Fix test case for test_apply_changelog_batch (#480)

faust/stores/rocksdb.py

@@ -414,7 +414,7 @@ def apply_changelog_batch(
         """
         batches: DefaultDict[int, WriteBatch]
         if self.use_rocksdict:
-            batches = defaultdict(lambda: WriteBatch(raw_mode=True))
+            batches = defaultdict(lambda: rocksdict.WriteBatch(raw_mode=True))
         else:
             batches = defaultdict(rocksdb.WriteBatch)
         tp_offsets: Dict[TP, int] = {}
@@ -424,13 +424,9 @@ def apply_changelog_batch(
                 offset if tp not in tp_offsets else max(offset, tp_offsets[tp])
             )
             msg = event.message
-            if self.use_rocksdict:
-                msg.key = msg.key.encode()
             if msg.value is None:
                 batches[msg.partition].delete(msg.key)
             else:
-                if self.use_rocksdict:
-                    msg.value = msg.value.encode()
                 batches[msg.partition].put(msg.key, msg.value)
 
         for partition, batch in batches.items():

tests/unit/stores/test_rocksdb.py

@@ -836,7 +836,6 @@ def test__itervalues(self, *, store):
             # TODO: seek_to_first() should be called once rocksdict is updated
             db.items.assert_called_once_with()
 
-    @pytest.mark.skip("Needs fixing")
     def test_apply_changelog_batch(self, *, store, rocksdict, db_for_partition):
         def new_event(name, tp: TP, offset, key, value) -> Mock:
             return Mock(