feat: Update ChangeSet to always require a Selector

ldclient/impl/datasource/feature_requester.py

@@ -13,7 +13,7 @@
 from ldclient.interfaces import FeatureRequester
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
-LATEST_ALL_URI = '/sdk/latest-all'
+FDV1_POLLING_ENDPOINT = '/sdk/latest-all'
 
 
 CacheEntry = namedtuple('CacheEntry', ['data', 'etag'])
@@ -24,7 +24,7 @@ def __init__(self, config):
         self._cache = dict()
         self._http = _http_factory(config).create_pool_manager(1, config.base_uri)
         self._config = config
-        self._poll_uri = config.base_uri + LATEST_ALL_URI
+        self._poll_uri = config.base_uri + FDV1_POLLING_ENDPOINT
         if config.payload_filter_key is not None:
             self._poll_uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key})
 

ldclient/impl/datasourcev2/polling.py

@@ -14,14 +14,13 @@
 import urllib3
 
 from ldclient.config import Config
-from ldclient.impl.datasource.feature_requester import LATEST_ALL_URI
+from ldclient.impl.datasource.feature_requester import FDV1_POLLING_ENDPOINT
 from ldclient.impl.datasystem.protocolv2 import (
     DeleteObject,
     EventName,
     PutObject
 )
 from ldclient.impl.http import _http_factory
-from ldclient.impl.repeating_task import RepeatingTask
 from ldclient.impl.util import (
     _LD_ENVID_HEADER,
     _LD_FD_FALLBACK_HEADER,
@@ -52,7 +51,7 @@
     Update
 )
 
-POLLING_ENDPOINT = "/sdk/poll"
+FDV2_POLLING_ENDPOINT = "/sdk/poll"
 
 
 PollingResult = _Result[Tuple[ChangeSet, Mapping], str]
@@ -95,9 +94,6 @@ def __init__(
         self._poll_interval = poll_interval
         self._interrupt_event = Event()
         self._stop = Event()
-        self._task = RepeatingTask(
-            "ldclient.datasource.polling", poll_interval, 0, self._poll
-        )
 
     @property
     def name(self) -> str:
@@ -193,7 +189,6 @@ def stop(self):
         """Stops the synchronizer."""
         log.info("Stopping PollingDataSourceV2 synchronizer")
         self._interrupt_event.set()
-        self._task.stop()
         self._stop.set()
 
     def _poll(self, ss: SelectorStore) -> BasisResult:
@@ -226,7 +221,7 @@ def _poll(self, ss: SelectorStore) -> BasisResult:
 
             basis = Basis(
                 change_set=change_set,
-                persist=change_set.selector is not None,
+                persist=change_set.selector.is_defined(),
                 environment_id=env_id,
             )
 
@@ -249,7 +244,7 @@ def __init__(self, config: Config):
         self._etag = None
         self._http = _http_factory(config).create_pool_manager(1, config.base_uri)
         self._config = config
-        self._poll_uri = config.base_uri + POLLING_ENDPOINT
+        self._poll_uri = config.base_uri + FDV2_POLLING_ENDPOINT
 
     def fetch(self, selector: Optional[Selector]) -> PollingResult:
         """
@@ -415,7 +410,7 @@ def __init__(self, config: Config):
         self._etag = None
         self._http = _http_factory(config).create_pool_manager(1, config.base_uri)
         self._config = config
-        self._poll_uri = config.base_uri + LATEST_ALL_URI
+        self._poll_uri = config.base_uri + FDV1_POLLING_ENDPOINT
 
     def fetch(self, selector: Optional[Selector]) -> PollingResult:
         """

ldclient/impl/datasourcev2/streaming.py

@@ -359,6 +359,7 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona
                     revert_to_fdv1=True,
                     environment_id=envid,
                 )
+                self.stop()
                 return (update, False)
 
             http_error_message_result = http_error_message(

ldclient/impl/datasystem/fdv2.py

@@ -414,7 +414,7 @@ def _run_initializers(self, set_on_ready: Event):
                 self._store.apply(basis.change_set, basis.persist)
 
                 # Set ready event if an only if a selector is defined for the changeset
-                if basis.change_set.selector is not None and basis.change_set.selector.is_defined():
+                if basis.change_set.selector.is_defined():
                     set_on_ready.set()
                     return
             except Exception as e:

ldclient/impl/datasystem/store.py

@@ -111,7 +111,6 @@ def apply_delta(self, collections: Collections) -> bool:
             with self._lock.write():
                 for kind, kind_data in all_decoded.items():
                     items_of_kind = self._items[kind]
-                    kind_data = all_decoded[kind]
                     for key, item in kind_data.items():
                         items_of_kind[key] = item
                         log.debug(
@@ -269,7 +268,7 @@ def apply(self, change_set: ChangeSet, persist: bool) -> None:
                 log.error("Store: couldn't apply changeset: %s", str(e))
 
     def _set_basis(
-        self, collections: Collections, selector: Optional[Selector], persist: bool
+        self, collections: Collections, selector: Selector, persist: bool
     ) -> None:
         """
         Set the basis of the store. Any existing data is discarded.
@@ -311,7 +310,7 @@ def _set_basis(
             self._send_change_events(affected_items)
 
     def _apply_delta(
-        self, collections: Collections, selector: Optional[Selector], persist: bool
+        self, collections: Collections, selector: Selector, persist: bool
     ) -> None:
         """
         Apply a delta update to the store.

ldclient/integrations/test_data.py

@@ -153,7 +153,7 @@ def __init__(self, key: str):
     # consider it part of the public API, but it is still called from TestData.
     def _copy(self) -> 'FlagBuilder':
         """Creates a deep copy of the flag builder. Subsequent updates to the
-        original ``FlagBuilder`` object will not update the copy and vise versa.
+        original ``FlagBuilder`` object will not update the copy and vice versa.
 
         :return: a copy of the flag builder object
         """

ldclient/integrations/test_datav2.py

@@ -182,7 +182,7 @@ def __init__(self, key: str):
     def _copy(self) -> FlagBuilderV2:
         """
         Creates a deep copy of the flag builder. Subsequent updates to the
-        original ``FlagBuilderV2`` object will not update the copy and vise versa.
+        original ``FlagBuilderV2`` object will not update the copy and vice versa.
 
         :return: a copy of the flag builder object
         """

ldclient/interfaces.py

@@ -1346,7 +1346,7 @@ class ChangeSet:
 
     intent_code: IntentCode
     changes: List[Change]
-    selector: Optional[Selector]
+    selector: Selector
 
 
 @dataclass(frozen=True)
@@ -1393,7 +1393,7 @@ def no_changes() -> "ChangeSet":
         require changes.
         """
         return ChangeSet(
-            intent_code=IntentCode.TRANSFER_NONE, selector=None, changes=[]
+            intent_code=IntentCode.TRANSFER_NONE, selector=Selector.no_selector(), changes=[]
         )
 
     @staticmethod

ldclient/testing/impl/datasourcev2/test_polling_initializer.py

@@ -7,6 +7,7 @@
     PollingDataSource,
     PollingResult,
     Selector,
+    fdv1_polling_payload_to_changeset,
     polling_payload_to_changeset
 )
 from ldclient.impl.util import UnsuccessfulResponseException, _Fail, _Success
@@ -138,3 +139,33 @@ def test_handles_transfer_changes():
     assert result.value.change_set.intent_code == IntentCode.TRANSFER_CHANGES
     assert len(result.value.change_set.changes) == 1
     assert result.value.persist is True
+
+
+def test_handles_fdv1_payload():
+    """Test that FDv1 payloads have persist set to False."""
+    fdv1_data = {
+        "flags": {
+            "test-flag": {
+                "key": "test-flag",
+                "version": 1,
+                "on": True,
+                "variations": [True, False]
+            }
+        },
+        "segments": {}
+    }
+    change_set_result = fdv1_polling_payload_to_changeset(fdv1_data)
+    assert isinstance(change_set_result, _Success)
+
+    mock_requester = MockPollingRequester(_Success(value=(change_set_result.value, {})))
+    ds = PollingDataSource(poll_interval=1.0, requester=mock_requester)
+
+    result = ds.fetch(MockSelectorStore(Selector.no_selector()))
+
+    assert isinstance(result, _Success)
+    assert result.value is not None
+
+    assert result.value.change_set.intent_code == IntentCode.TRANSFER_FULL
+    assert len(result.value.change_set.changes) == 1
+    # FDv1 payloads should not be persisted because they have no selector
+    assert result.value.persist is False

ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py

@@ -46,7 +46,7 @@ def test_transfer_none():
     change_set = result.value
     assert change_set.intent_code == IntentCode.TRANSFER_NONE
     assert len(change_set.changes) == 0
-    assert change_set.selector is None
+    assert not change_set.selector.is_defined()
 
 
 def test_transfer_full_with_empty_payload():
@@ -58,7 +58,7 @@ def test_transfer_full_with_empty_payload():
     change_set = result.value
     assert change_set.intent_code == IntentCode.TRANSFER_FULL
     assert len(change_set.changes) == 0
-    assert change_set.selector is not None
+    assert change_set.selector.is_defined()
     assert change_set.selector.state == "(p:5A46PZ79FQ9D08YYKT79DECDNV:461)"
     assert change_set.selector.version == 461
 
@@ -86,7 +86,7 @@ def test_processes_put_object():
     assert change_set.changes[0].version == 461
     assert isinstance(change_set.changes[0].object, dict)
 
-    assert change_set.selector is not None
+    assert change_set.selector.is_defined()
     assert change_set.selector.state == "(p:5A46PZ79FQ9D08YYKT79DECDNV:461)"
     assert change_set.selector.version == 461
 
@@ -106,7 +106,7 @@ def test_processes_delete_object():
     assert change_set.changes[0].version == 461
     assert change_set.changes[0].object is None
 
-    assert change_set.selector is not None
+    assert change_set.selector.is_defined()
     assert change_set.selector.state == "(p:5A46PZ79FQ9D08YYKT79DECDNV:461)"
     assert change_set.selector.version == 461
 
@@ -168,7 +168,6 @@ def test_fdv1_payload_empty_flags_and_segments():
     assert change_set.intent_code == IntentCode.TRANSFER_FULL
     assert len(change_set.changes) == 0
     # FDv1 doesn't use selectors
-    assert change_set.selector is not None
     assert not change_set.selector.is_defined()
 
 

ldclient/testing/impl/datasourcev2/test_polling_synchronizer.py

@@ -142,7 +142,7 @@ def test_handles_empty_changeset():
 
     assert valid.change_set is not None
     assert len(valid.change_set.changes) == 0
-    assert valid.change_set.selector is not None
+    assert valid.change_set.selector.is_defined()
     assert valid.change_set.selector.version == 300
     assert valid.change_set.selector.state == "p:SOMETHING:300"
     assert valid.change_set.intent_code == IntentCode.TRANSFER_FULL
@@ -175,7 +175,7 @@ def test_handles_put_objects():
     assert valid.change_set.changes[0].key == "flag-key"
     assert valid.change_set.changes[0].object == {"key": "flag-key"}
     assert valid.change_set.changes[0].version == 100
-    assert valid.change_set.selector is not None
+    assert valid.change_set.selector.is_defined()
     assert valid.change_set.selector.version == 300
     assert valid.change_set.selector.state == "p:SOMETHING:300"
     assert valid.change_set.intent_code == IntentCode.TRANSFER_FULL
@@ -205,7 +205,7 @@ def test_handles_delete_objects():
     assert valid.change_set.changes[0].kind == ObjectKind.FLAG
     assert valid.change_set.changes[0].key == "flag-key"
     assert valid.change_set.changes[0].version == 101
-    assert valid.change_set.selector is not None
+    assert valid.change_set.selector.is_defined()
     assert valid.change_set.selector.version == 300
     assert valid.change_set.selector.state == "p:SOMETHING:300"
     assert valid.change_set.intent_code == IntentCode.TRANSFER_FULL

ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py

@@ -209,7 +209,7 @@ def test_handles_empty_changeset(events):  # pylint: disable=redefined-outer-nam
 
     assert updates[0].change_set is not None
     assert len(updates[0].change_set.changes) == 0
-    assert updates[0].change_set.selector is not None
+    assert updates[0].change_set.selector.is_defined()
     assert updates[0].change_set.selector.version == 300
     assert updates[0].change_set.selector.state == "p:SOMETHING:300"
     assert updates[0].change_set.intent_code == IntentCode.TRANSFER_FULL
@@ -241,7 +241,7 @@ def test_handles_put_objects(events):  # pylint: disable=redefined-outer-name
     assert updates[0].change_set.changes[0].key == "flag-key"
     assert updates[0].change_set.changes[0].object == {"key": "flag-key"}
     assert updates[0].change_set.changes[0].version == 100
-    assert updates[0].change_set.selector is not None
+    assert updates[0].change_set.selector.is_defined()
     assert updates[0].change_set.selector.version == 300
     assert updates[0].change_set.selector.state == "p:SOMETHING:300"
     assert updates[0].change_set.intent_code == IntentCode.TRANSFER_FULL
@@ -272,7 +272,7 @@ def test_handles_delete_objects(events):  # pylint: disable=redefined-outer-name
     assert updates[0].change_set.changes[0].kind == ObjectKind.FLAG
     assert updates[0].change_set.changes[0].key == "flag-key"
     assert updates[0].change_set.changes[0].version == 101
-    assert updates[0].change_set.selector is not None
+    assert updates[0].change_set.selector.is_defined()
     assert updates[0].change_set.selector.version == 300
     assert updates[0].change_set.selector.state == "p:SOMETHING:300"
     assert updates[0].change_set.intent_code == IntentCode.TRANSFER_FULL
@@ -299,7 +299,7 @@ def test_swallows_goodbye(events):  # pylint: disable=redefined-outer-name
 
     assert updates[0].change_set is not None
     assert len(updates[0].change_set.changes) == 0
-    assert updates[0].change_set.selector is not None
+    assert updates[0].change_set.selector.is_defined()
     assert updates[0].change_set.selector.version == 300
     assert updates[0].change_set.selector.state == "p:SOMETHING:300"
     assert updates[0].change_set.intent_code == IntentCode.TRANSFER_FULL
@@ -326,7 +326,7 @@ def test_swallows_heartbeat(events):  # pylint: disable=redefined-outer-name
 
     assert updates[0].change_set is not None
     assert len(updates[0].change_set.changes) == 0
-    assert updates[0].change_set.selector is not None
+    assert updates[0].change_set.selector.is_defined()
     assert updates[0].change_set.selector.version == 300
     assert updates[0].change_set.selector.state == "p:SOMETHING:300"
     assert updates[0].change_set.intent_code == IntentCode.TRANSFER_FULL

ldclient/testing/impl/datasystem/test_fdv2_persistence.py

@@ -370,7 +370,8 @@ def test_persistent_store_delete_operations():
         ChangeSet,
         ChangeType,
         IntentCode,
-        ObjectKind
+        ObjectKind,
+        Selector
     )
 
     # Pre-populate with a flag
@@ -410,7 +411,7 @@ def test_persistent_store_delete_operations():
                 },
             )
         ],
-        selector=None,
+        selector=Selector.no_selector(),
     )
     store.apply(init_changeset, True)
 
@@ -428,7 +429,7 @@ def test_persistent_store_delete_operations():
                 object=None,
             )
         ],
-        selector=None,
+        selector=Selector.no_selector(),
     )
 
     store.apply(delete_changeset, True)
@@ -671,7 +672,8 @@ def test_persistent_store_commit_encodes_data_correctly():
         ChangeSet,
         ChangeType,
         IntentCode,
-        ObjectKind
+        ObjectKind,
+        Selector
     )
 
     persistent_store = StubFeatureStore()
@@ -699,7 +701,7 @@ def test_persistent_store_commit_encodes_data_correctly():
                 object=flag_data,
             )
         ],
-        selector=None,
+        selector=Selector.no_selector(),
     )
     store.apply(changeset, True)
 
@@ -746,7 +748,8 @@ def test_persistent_store_commit_handles_errors():
         ChangeSet,
         ChangeType,
         IntentCode,
-        ObjectKind
+        ObjectKind,
+        Selector
     )
 
     class FailingFeatureStore(StubFeatureStore):
@@ -770,7 +773,7 @@ def init(self, all_data):
                 object={"key": "test-flag", "version": 1, "on": True},
             )
         ],
-        selector=None,
+        selector=Selector.no_selector(),
     )
     store.apply(changeset, True)