|
| 1 | +import importlib.util |
| 2 | +import unittest |
| 3 | +from unittest.mock import Mock, patch |
| 4 | + |
| 5 | +from datadog_lambda import feature_flags |
| 6 | +from datadog_lambda import wrapper |
| 7 | + |
| 8 | + |
| 9 | +class FeatureFlagFlushTest(unittest.TestCase): |
| 10 | + def test_budget(self): |
| 11 | + for remaining, cap, expected in [ |
| 12 | + (5000, 500, 0.5), |
| 13 | + (200, 500, 0.1), |
| 14 | + (101, 500, 0.001), |
| 15 | + ]: |
| 16 | + with self.subTest(remaining=remaining), patch.object( |
| 17 | + feature_flags, "_telemetry" |
| 18 | + ) as telemetry: |
| 19 | + telemetry.FLUSH_API_VERSION = 1 |
| 20 | + telemetry.flush.return_value = "completed" |
| 21 | + with patch.object( |
| 22 | + type(feature_flags.config), "feature_flag_flush_timeout_ms", cap |
| 23 | + ): |
| 24 | + feature_flags.flush_feature_flag_telemetry( |
| 25 | + Mock(get_remaining_time_in_millis=lambda: remaining) |
| 26 | + ) |
| 27 | + telemetry.flush.assert_called_once_with(expected) |
| 28 | + |
| 29 | + def test_no_budget(self): |
| 30 | + for remaining, cap in [ |
| 31 | + (100, 500), |
| 32 | + (0, 500), |
| 33 | + (1000, 0), |
| 34 | + (1000, -1), |
| 35 | + (1000, float("nan")), |
| 36 | + ]: |
| 37 | + with self.subTest(remaining=remaining, cap=cap), patch.object( |
| 38 | + feature_flags, "_telemetry" |
| 39 | + ) as telemetry: |
| 40 | + telemetry.FLUSH_API_VERSION = 1 |
| 41 | + with patch.object( |
| 42 | + type(feature_flags.config), "feature_flag_flush_timeout_ms", cap |
| 43 | + ): |
| 44 | + feature_flags.flush_feature_flag_telemetry( |
| 45 | + Mock(get_remaining_time_in_millis=lambda: remaining) |
| 46 | + ) |
| 47 | + telemetry.flush.assert_not_called() |
| 48 | + |
| 49 | + def test_old_tracer(self): |
| 50 | + for telemetry in [None, Mock(FLUSH_API_VERSION=2)]: |
| 51 | + with patch.object(feature_flags, "_telemetry", telemetry): |
| 52 | + feature_flags.flush_feature_flag_telemetry(None) |
| 53 | + |
| 54 | + def test_missing_context_and_telemetry_errors_are_safe(self): |
| 55 | + with patch.object(feature_flags, "_telemetry") as telemetry: |
| 56 | + telemetry.FLUSH_API_VERSION = 1 |
| 57 | + telemetry.flush.side_effect = RuntimeError("unavailable") |
| 58 | + feature_flags.flush_feature_flag_telemetry(None) |
| 59 | + feature_flags.flush_feature_flag_telemetry( |
| 60 | + Mock(get_remaining_time_in_millis=lambda: 1000) |
| 61 | + ) |
| 62 | + telemetry.flush.assert_called_once() |
| 63 | + |
| 64 | + def test_incomplete_diagnostic(self): |
| 65 | + with patch.object(feature_flags, "_telemetry") as telemetry, patch.object( |
| 66 | + feature_flags, "logger" |
| 67 | + ) as log: |
| 68 | + telemetry.FLUSH_API_VERSION = 1 |
| 69 | + telemetry.flush.return_value = "timed_out" |
| 70 | + feature_flags.flush_feature_flag_telemetry( |
| 71 | + Mock(get_remaining_time_in_millis=lambda: 1000) |
| 72 | + ) |
| 73 | + log.debug.assert_called_with( |
| 74 | + "Feature flag telemetry flush incomplete: %s", "timed_out" |
| 75 | + ) |
| 76 | + |
| 77 | + def test_wrapper_preserves_result_and_exception_when_cleanup_fails(self): |
| 78 | + for raises in [False, True]: |
| 79 | + original = ValueError("handler error") |
| 80 | + handler = ( |
| 81 | + Mock(side_effect=original) |
| 82 | + if raises |
| 83 | + else Mock(return_value={"ok": True}) |
| 84 | + ) |
| 85 | + # Isolate invocation control flow from tracing/metrics initialization. |
| 86 | + decorated = object.__new__(wrapper._LambdaDecorator) |
| 87 | + decorated.func = handler |
| 88 | + decorated.blocking_response = None |
| 89 | + decorated.span = None |
| 90 | + decorated._before = Mock() |
| 91 | + decorated._after = Mock(side_effect=RuntimeError("cleanup error")) |
| 92 | + context = Mock(get_remaining_time_in_millis=lambda: 30000) |
| 93 | + with patch.object(wrapper, "flush_feature_flag_telemetry") as flush, patch( |
| 94 | + "datadog_lambda.metric.submit_errors_metric", |
| 95 | + side_effect=RuntimeError("metrics error"), |
| 96 | + ): |
| 97 | + if raises: |
| 98 | + with self.assertRaises(ValueError) as caught: |
| 99 | + decorated({}, context) |
| 100 | + self.assertIs(caught.exception, original) |
| 101 | + else: |
| 102 | + self.assertEqual(decorated({}, context), {"ok": True}) |
| 103 | + flush.assert_called_once_with(context) |
| 104 | + |
| 105 | + |
| 106 | +class FeatureFlagTracerIntegrationTest(unittest.TestCase): |
| 107 | + @unittest.skipUnless( |
| 108 | + feature_flags._telemetry is not None |
| 109 | + and importlib.util.find_spec("openfeature") is not None, |
| 110 | + "requires tracer flush contract v1 and the optional OpenFeature SDK", |
| 111 | + ) |
| 112 | + def test_short_warm_invocations_flush_both_writers(self): |
| 113 | + from ddtrace.internal.openfeature._flagevaluation_writer import ( |
| 114 | + FlagEvaluationWriter, |
| 115 | + _EvalEvent, |
| 116 | + ) |
| 117 | + from ddtrace.internal.openfeature._evp_transport import EVPRoute |
| 118 | + from ddtrace.internal.openfeature.writer import ExposureWriter |
| 119 | + |
| 120 | + route = EVPRoute("http://agent:8126", "/evp_proxy/v2", {}) |
| 121 | + selector = Mock(configuration_source="agentless") |
| 122 | + selector.select.return_value = route |
| 123 | + selector.send.side_effect = lambda route, send: send(route) |
| 124 | + connection = Mock() |
| 125 | + connection.getresponse.return_value.status = 202 |
| 126 | + evaluations = FlagEvaluationWriter( |
| 127 | + route_selector=selector, connection_factory=lambda *a, **k: connection |
| 128 | + ) |
| 129 | + exposures = ExposureWriter( |
| 130 | + enabled=True, |
| 131 | + route_selector=selector, |
| 132 | + connection_factory=lambda *a, **k: connection, |
| 133 | + ) |
| 134 | + original = ValueError("handler failure") |
| 135 | + |
| 136 | + def handler(event, context): |
| 137 | + evaluations.enqueue( |
| 138 | + _EvalEvent("flag", "on", "allocation", "subject", {}, False, "", 1) |
| 139 | + ) |
| 140 | + exposures.enqueue({"timestamp": 1}) |
| 141 | + if event.get("raise"): |
| 142 | + raise original |
| 143 | + return {"ok": True} |
| 144 | + |
| 145 | + decorated = object.__new__(wrapper._LambdaDecorator) |
| 146 | + decorated.func = handler |
| 147 | + decorated.blocking_response = None |
| 148 | + decorated.span = None |
| 149 | + decorated._before = Mock() |
| 150 | + decorated._after = Mock(side_effect=RuntimeError("unrelated cleanup")) |
| 151 | + context = Mock(get_remaining_time_in_millis=lambda: 30000) |
| 152 | + with patch("datadog_lambda.metric.submit_errors_metric"): |
| 153 | + self.assertEqual(decorated({}, context), {"ok": True}) |
| 154 | + self.assertEqual(connection.request.call_count, 2) |
| 155 | + with self.assertRaises(ValueError) as caught: |
| 156 | + decorated({"raise": True}, context) |
| 157 | + self.assertIs(caught.exception, original) |
| 158 | + self.assertEqual(connection.request.call_count, 4) |
| 159 | + self.assertEqual(decorated({}, context), {"ok": True}) |
| 160 | + self.assertEqual(connection.request.call_count, 6) |
0 commit comments