add: test cases and reconnect functionality

Signed-off-by: Cole Bailey <[email protected]>

.github/workflows/build.yml

@@ -27,6 +27,28 @@ jobs:
           - "hooks/openfeature-hooks-opentelemetry"
           - "providers/openfeature-provider-flagd"
 
+    services:
+      # flagd-testbed for flagd RPC provider e2e tests
+      flagd:
+        image: ghcr.io/open-feature/flagd-testbed:v0.5.4
+        ports:
+          - 8013:8013
+      # flagd-testbed for flagd RPC provider reconnect e2e tests
+      flagd-unstable:
+        image: ghcr.io/open-feature/flagd-testbed-unstable:v0.5.4
+        ports:
+          - 8014:8013
+      # sync-testbed for flagd in-process provider e2e tests
+      sync:
+        image: ghcr.io/open-feature/sync-testbed:v0.5.4
+        ports:
+          - 9090:9090
+      # sync-testbed for flagd in-process provider reconnect e2e tests
+      sync-unstable:
+        image: ghcr.io/open-feature/sync-testbed-unstable:v0.5.4
+        ports:
+          - 9091:9090
+
     steps:
       - uses: actions/checkout@v4
         with:

providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/file_watcher.py

@@ -30,6 +30,7 @@ def __init__(
         self.last_modified = 0.0
         self.flag_data: typing.Mapping[str, Flag] = {}
         self.load_data()
+        self.has_error = False
         self.thread = threading.Thread(target=self.refresh_file, daemon=True)
         self.thread.start()
 
@@ -57,17 +58,31 @@ def load_data(self, modified_time: typing.Optional[float] = None) -> None:
 
                 self.flag_data = Flag.parse_flags(data)
                 logger.debug(f"{self.flag_data=}")
+
+                if self.has_error:
+                    self.provider.emit_provider_ready(
+                        ProviderEventDetails(
+                            message="Reloading file contents recovered from error state"
+                        )
+                    )
+                    self.has_error = False
+
                 self.provider.emit_provider_configuration_changed(
                     ProviderEventDetails(flags_changed=list(self.flag_data.keys()))
                 )
             self.last_modified = modified_time or os.path.getmtime(self.file_path)
         except FileNotFoundError:
-            logger.exception("Provided file path not valid")
+            self.handle_error("Provided file path not valid")
         except json.JSONDecodeError:
-            logger.exception("Could not parse JSON flag data from file")
+            self.handle_error("Could not parse JSON flag data from file")
         except yaml.error.YAMLError:
-            logger.exception("Could not parse YAML flag data from file")
+            self.handle_error("Could not parse YAML flag data from file")
         except ParseError:
-            logger.exception("Could not parse flag data using flagd syntax")
+            self.handle_error("Could not parse flag data using flagd syntax")
         except Exception:
-            logger.exception("Could not read flags from file")
+            self.handle_error("Could not read flags from file")
+
+    def handle_error(self, error_message: str) -> None:
+        logger.exception(error_message)
+        self.has_error = True
+        self.provider.emit_provider_error(ProviderEventDetails(message=error_message))

providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/grpc_watcher.py

@@ -7,7 +7,7 @@
 import grpc
 
 from openfeature.event import ProviderEventDetails
-from openfeature.exception import ParseError
+from openfeature.exception import ErrorCode, ParseError
 from openfeature.provider.provider import AbstractProvider
 
 from ....config import Config
@@ -68,13 +68,19 @@ def sync_flags(self) -> None:
                     )
                     self.flag_data = Flag.parse_flags(json.loads(flag_str))
 
-                    self.connected = True
+                    if not self.connected:
+                        self.provider.emit_provider_ready(
+                            ProviderEventDetails(
+                                message="gRPC sync connection established"
+                            )
+                        )
+                        self.connected = True
+                        # reset retry delay after successsful read
+                        retry_delay = self.INIT_BACK_OFF
+
                     self.provider.emit_provider_configuration_changed(
                         ProviderEventDetails(flags_changed=list(self.flag_data.keys()))
                     )
-
-                    # reset retry delay after successsful read
-                    retry_delay = self.INIT_BACK_OFF
             except grpc.RpcError as e:  # noqa: PERF203
                 logger.error(f"SyncFlags stream error, {e.code()=} {e.details()=}")
             except json.JSONDecodeError:
@@ -86,7 +92,13 @@ def sync_flags(self) -> None:
                     f"Could not parse flag data using flagd syntax: {flag_str=}"
                 )
             finally:
-                # self.connected = False
+                self.connected = False
+                self.provider.emit_provider_error(
+                    ProviderEventDetails(
+                        message=f"gRPC sync disconnected, reconnecting in {retry_delay}s",
+                        error_code=ErrorCode.GENERAL,
+                    )
+                )
                 logger.info(f"Reconnecting in {retry_delay}s")
                 time.sleep(retry_delay)
                 retry_delay = min(2 * retry_delay, self.MAX_BACK_OFF)

providers/openfeature-provider-flagd/tests/e2e/conftest.py

@@ -1,9 +1,10 @@
+import time
 import typing
 
 import pytest
 from pytest_bdd import parsers, then, when
 
-from openfeature.client import OpenFeatureClient
+from openfeature.client import OpenFeatureClient, ProviderEvent
 from openfeature.evaluation_context import EvaluationContext
 
 JsonPrimitive = typing.Union[str, bool, float, int]
@@ -194,3 +195,112 @@ def assert_reason(
     key, default = key_and_default
     evaluation_result = client.get_string_details(key, default, evaluation_context)
     assert evaluation_result.reason.value == reason
+
+
+@pytest.fixture
+def handles() -> list:
+    return []
+
+
+@when(
+    parsers.cfparse(
+        "a {event_type:ProviderEvent} handler is added",
+        extra_types={"ProviderEvent": ProviderEvent},
+    ),
+    target_fixture="handles",
+)
+def add_event_handler(
+    client: OpenFeatureClient, event_type: ProviderEvent, handles: list
+):
+    def handler(event):
+        handles.append(
+            {
+                "type": event_type,
+                "event": event,
+            }
+        )
+
+    client.add_handler(event_type, handler)
+    return handles
+
+
+@when(
+    parsers.cfparse(
+        "a {event_type:ProviderEvent} handler and a {event_type2:ProviderEvent} handler are added",
+        extra_types={"ProviderEvent": ProviderEvent},
+    ),
+    target_fixture="handles",
+)
+def add_event_handlers(
+    client: OpenFeatureClient,
+    event_type: ProviderEvent,
+    event_type2: ProviderEvent,
+    handles: list,
+):
+    add_event_handler(client, event_type, handles)
+    add_event_handler(client, event_type2, handles)
+
+
+def assert_handlers(
+    handles, event_type: ProviderEvent, max_wait: int = 2, num_events: int = 1
+):
+    poll_interval = 0.05
+    while max_wait > 0:
+        if len([h["type"] == event_type for h in handles]) < num_events:
+            max_wait -= poll_interval
+            time.sleep(poll_interval)
+            continue
+        break
+
+    actual_num_events = len([h["type"] == event_type for h in handles])
+    assert (
+        num_events <= actual_num_events
+    ), f"Expected {num_events} but got {actual_num_events}: {handles}"
+
+
+@then(
+    parsers.cfparse(
+        "the {event_type:ProviderEvent} handler must run",
+        extra_types={"ProviderEvent": ProviderEvent},
+    )
+)
+@then(
+    parsers.cfparse(
+        "the {event_type:ProviderEvent} handler must run when the provider connects",
+        extra_types={"ProviderEvent": ProviderEvent},
+    )
+)
+def assert_handler_run(handles, event_type: ProviderEvent):
+    assert_handlers(handles, event_type, max_wait=3)
+
+
+@then(
+    parsers.cfparse(
+        "the {event_type:ProviderEvent} handler must run when the provider's connection is lost",
+        extra_types={"ProviderEvent": ProviderEvent},
+    )
+)
+def assert_disconnect_handler(handles, event_type: ProviderEvent):
+    assert_handlers(handles, event_type, max_wait=6)
+
+
+@then(
+    parsers.cfparse(
+        "when the connection is reestablished the {event_type:ProviderEvent} handler must run again",
+        extra_types={"ProviderEvent": ProviderEvent},
+    )
+)
+def assert_disconnect_error(handles, event_type: ProviderEvent):
+    assert_handlers(handles, event_type, max_wait=6, num_events=2)
+
+
+@then(parsers.cfparse('the event details must indicate "{key}" was altered'))
+def assert_flag_changed(handles, key):
+    handle = None
+    for h in handles:
+        if h["type"] == ProviderEvent.PROVIDER_CONFIGURATION_CHANGED:
+            handle = h
+            break
+
+    assert handle is not None
+    assert key in handle["event"].flags_changed

providers/openfeature-provider-flagd/tests/e2e/inprocess/file/test_inprocess_events.py

@@ -2,11 +2,9 @@
 import time
 
 import pytest
-from pytest_bdd import parsers, scenario, then, when
+from pytest_bdd import parsers, scenario, when
 from tests.conftest import setup_flag_file
 
-from openfeature.client import OpenFeatureClient, ProviderEvent
-
 GHERKIN_FOLDER = "../../../../test-harness/gherkin/"
 
 
@@ -26,68 +24,10 @@ def flag_file(tmp_path):
     return setup_flag_file(tmp_path, "changing-flag-bar.json")
 
 
-@pytest.fixture
-def handles() -> list:
-    return []
-
-
-@when(
-    parsers.cfparse(
-        "a {event_type:ProviderEvent} handler is added",
-        extra_types={"ProviderEvent": ProviderEvent},
-    ),
-    target_fixture="handles",
-)
-def add_event_handler(
-    client: OpenFeatureClient, event_type: ProviderEvent, handles: list
-):
-    def handler(event):
-        handles.append(
-            {
-                "type": event_type,
-                "event": event,
-            }
-        )
-
-    client.add_handler(event_type, handler)
-    return handles
-
-
-@then(
-    parsers.cfparse(
-        "the {event_type:ProviderEvent} handler must run",
-        extra_types={"ProviderEvent": ProviderEvent},
-    )
-)
-def assert_handler_run(handles, event_type: ProviderEvent):
-    max_wait = 2
-    poll_interval = 0.1
-    while max_wait > 0:
-        if all(h["type"] != event_type for h in handles):
-            max_wait -= poll_interval
-            time.sleep(poll_interval)
-            continue
-        break
-
-    assert any(h["type"] == event_type for h in handles)
-
-
 @when(parsers.cfparse('a flag with key "{key}" is modified'))
 def modify_flag(flag_file, key):
     time.sleep(0.1)  # guard against race condition
     with open("test-harness/flags/changing-flag-foo.json") as src_file:
         contents = src_file.read()
     with open(flag_file, "w") as f:
         f.write(contents)
-
-
-@then(parsers.cfparse('the event details must indicate "{key}" was altered'))
-def assert_flag_changed(handles, key):
-    handle = None
-    for h in handles:
-        if h["type"] == ProviderEvent.PROVIDER_CONFIGURATION_CHANGED:
-            handle = h
-            break
-
-    assert handle is not None
-    assert key in handle["event"].flags_changed

providers/openfeature-provider-flagd/tests/e2e/inprocess/grpc/conftest.py

@@ -1,8 +1,9 @@
 import pytest
-from pytest_bdd import given
+from pytest_bdd import given, parsers, then, when
+from tests.e2e.conftest import add_event_handler, assert_handlers
 
 from openfeature import api
-from openfeature.client import OpenFeatureClient
+from openfeature.client import OpenFeatureClient, ProviderEvent
 from openfeature.contrib.provider.flagd import FlagdProvider
 from openfeature.contrib.provider.flagd.config import ResolverType
 
@@ -22,3 +23,25 @@ def setup_provider(port: int) -> OpenFeatureClient:
         )
     )
     return api.get_client()
+
+
+@when(parsers.cfparse('a flag with key "{key}" is modified'))
+def modify_flag(key):
+    # sync service will flip flag contents regularly
+    pass
+
+
+@given("flagd is unavailable", target_fixture="client")
+def flagd_unavailable():
+    return setup_provider(99999)
+
+
+@when("a flagd provider is set and initialization is awaited")
+def flagd_init(client: OpenFeatureClient, handles):
+    add_event_handler(client, ProviderEvent.PROVIDER_ERROR, handles)
+    add_event_handler(client, ProviderEvent.PROVIDER_READY, handles)
+
+
+@then("an error should be indicated within the configured deadline")
+def flagd_error(handles):
+    assert_handlers(handles, ProviderEvent.PROVIDER_ERROR)