featured: update and extend tests for deregister race guards

test_feature_resync: the sub-case where get_entry returns None previously
asserted that mod_entry would be called with the rendered state.  That
behaviour was the root of the bug -- resync_feature_state was recreating a
FEATURE row that had been deleted by a concurrent deregister.  Update the
assertion to assert_not_called(), matching the new guard that returns early
when the entry is absent.  Add a further sub-case for a partial entry (row
exists but has no 'state' key, the TOCTOU artifact left by sync_feature_scope
racing with deregister): resync_feature_state must call set_entry(None) to
remove the ghost record rather than completing it with mod_entry.

test_sync_feature_scope_toctou: new test for the post-write TOCTOU guard in
sync_feature_scope.  Uses side_effect to return a live entry on the pre-write
check and a partial entry (no 'state' key) on the post-write check, simulating
the window where the FEATURE row is deleted and recreated by mod_entry during
the race.  sync_feature_scope must call set_entry(None) to clean it up.

Signed-off-by: david.zagury <davidza@nvidia.com>

Author: DavidZagury

scripts/featured

@@ -353,6 +353,7 @@ class FeatureHandler(object):
         new_per_asic = str(feature_config.has_per_asic_scope)
         new_global = str(feature_config.has_global_scope)
         self._conditional_update_scope(self._config_db, feature_config.name, new_per_asic, new_global)
+
         # If the entry was deleted by a concurrent deregister, _conditional_update_scope
         # may have recreated it without a 'state' field.  Remove the partial entry to
         # avoid a ghost record.

tests/featured/featured_test.py

@@ -282,7 +282,8 @@ def test_feature_resync(self):
         }
         mock_db.get_entry.return_value = None
         feature_handler.sync_state_field(feature_table)
-        mock_db.mod_entry.assert_called_with('FEATURE', 'sflow', {'state': 'enabled'})
+        # Missing FEATURE row: resync_feature_state skips mod_entry (avoid recreating a row removed concurrently).
+        mock_db.mod_entry.assert_not_called()
         mock_db.mod_entry.reset_mock()
 
         feature_handler = featured.FeatureHandler(mock_db, mock_feature_state_table, {}, False)
@@ -321,7 +322,46 @@ def test_feature_resync(self):
         }
         feature_handler.sync_state_field(feature_table)
         mock_db.mod_entry.assert_called_with('FEATURE', 'sflow', {'state': 'enabled'})
-    
+        mock_db.mod_entry.reset_mock()
+
+        # Partial entry (no 'state' key): left behind by sync_feature_scope racing with
+        # deregister.  resync_feature_state must delete it rather than completing it.
+        feature_handler = featured.FeatureHandler(mock_db, mock_feature_state_table, {}, False)
+        mock_db.get_entry.return_value = {
+            'has_per_asic_scope': 'False',
+            'has_global_scope': 'True',
+        }
+        mock_db.set_entry = mock.MagicMock()
+        feature_handler.sync_state_field(feature_table)
+        mock_db.set_entry.assert_called_with('FEATURE', 'sflow', None)
+        mock_db.mod_entry.assert_not_called()
+
+    def test_sync_feature_scope_toctou(self):
+        """sync_feature_scope must delete a partial entry it accidentally creates when
+        the FEATURE row is deleted by a concurrent deregister between the mod_entry
+        calls and the post-write existence check (TOCTOU)."""
+        mock_db = mock.MagicMock()
+        mock_feature_state_table = mock.MagicMock()
+
+        feature_handler = featured.FeatureHandler(mock_db, mock_feature_state_table, {}, False)
+        feature = featured.Feature('sflow', {
+            'state': 'disabled',
+            'has_global_scope': 'True',
+            'has_per_asic_scope': 'False',
+            'delayed': 'False',
+        })
+
+        # Post-write check sees a partial entry (no 'state' key): the FEATURE row was
+        # deleted by deregister and then recreated by mod_entry during the race
+        # window, leaving only the scope fields.
+        mock_db.get_entry.return_value = {
+            'has_per_asic_scope': 'False',
+            'has_global_scope': 'True',
+        }
+
+        feature_handler.sync_feature_scope(feature)
+        mock_db.set_entry.assert_called_with(featured.FEATURE_TBL, 'sflow', None)
+
     def test_port_init_done_twice(self):
         """There could be multiple "PortInitDone" event in case of swss
         restart(either due to crash or due to manual operation). swss