preferences: refactor cf and sc

src/rpcclient/rpcclient/darwin/cfpreferences.py

@@ -20,36 +20,50 @@ def __init__(self, client):
         """
         self._client = client
 
-    def copy_key_list(self, application_id: str, username: str = kCFPreferencesCurrentUser,
-                      hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[typing.List[str]]:
+    def get_keys(self, application_id: str, username: str = kCFPreferencesCurrentUser,
+                 hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[typing.List[str]]:
         application_id = self._client.cf(application_id)
         username = self._client.cf(username)
         hostname = self._client.cf(hostname)
         return self._client.symbols.CFPreferencesCopyKeyList(application_id, username, hostname).py
 
-    def copy_value(self, key: str, application_id: str, username: str = kCFPreferencesCurrentUser,
-                   hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[str]:
+    def get_value(self, key: str, application_id: str, username: str = kCFPreferencesCurrentUser,
+                  hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[str]:
         key = self._client.cf(key)
         application_id = self._client.cf(application_id)
         username = self._client.cf(username)
         hostname = self._client.cf(hostname)
         return self._client.symbols.CFPreferencesCopyValue(key, application_id, username, hostname).py
 
-    def copy_all_values(self, application_id: str, username: str = kCFPreferencesCurrentUser,
-                        hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[typing.Mapping]:
+    def get_values(self, application_id: str, username: str = kCFPreferencesCurrentUser,
+                   hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[typing.Mapping]:
         result = {}
-        key_list = self.copy_key_list(application_id, username, hostname)
+        key_list = self.get_keys(application_id, username, hostname)
         if not key_list:
             raise RpcClientException(f'failed to get key list for: {application_id}/{username}/{hostname}')
         for k in key_list:
-            result[k] = self.copy_value(k, application_id, username, hostname)
+            result[k] = self.get_value(k, application_id, username, hostname)
         return result
 
-    def set_value(self, key: str, value: str, application_id: str, username: str = kCFPreferencesCurrentUser,
-                  hostname: str = kCFPreferencesCurrentHost):
-        key = self._client.cf(key)
-        value = self._client.cf(value)
-        application_id = self._client.cf(application_id)
-        username = self._client.cf(username)
-        hostname = self._client.cf(hostname)
-        self._client.symbols.CFPreferencesSetValue(key, value, application_id, username, hostname)
+    def set(self, key: str, value: str, application_id: str, username: str = kCFPreferencesCurrentUser,
+            hostname: str = kCFPreferencesCurrentHost):
+        self._client.symbols.CFPreferencesSetValue(self._client.cf(key), self._client.cf(value),
+                                                   self._client.cf(application_id), self._client.cf(username),
+                                                   self._client.cf(hostname))
+
+    def remove(self, key: str, application_id: str, username: str = kCFPreferencesCurrentUser,
+               hostname: str = kCFPreferencesCurrentHost):
+        self._client.symbols.CFPreferencesSetValue(self._client.cf(key), 0,
+                                                   self._client.cf(application_id), self._client.cf(username),
+                                                   self._client.cf(hostname))
+
+    def set_dict(self, d: typing.Mapping, application_id: str, username: str = kCFPreferencesCurrentUser,
+                 hostname: str = kCFPreferencesCurrentHost):
+        for k, v in d.items():
+            self.set(k, v, application_id, username, hostname)
+
+    def clear(self, application_id: str, username: str = kCFPreferencesCurrentUser,
+              hostname: str = kCFPreferencesCurrentHost):
+        """ remove all values from given preference """
+        for k in self.get_keys(application_id, username, hostname):
+            self.remove(k, application_id, username, hostname)

src/rpcclient/rpcclient/darwin/client.py

@@ -41,6 +41,7 @@ def __init__(self, sock, sysname: str, hostname: str, port: int = None):
             raise MissingLibraryError('failed to load CoreFoundation')
 
         self._cf_types = {
+            self.symbols.CFNullGetTypeID(): 'null',
             self.symbols.CFDateGetTypeID(): 'date',
             self.symbols.CFDataGetTypeID(): 'data',
             self.symbols.CFStringGetTypeID(): 'str',

src/rpcclient/rpcclient/darwin/scpreferences.py

@@ -19,6 +19,10 @@ def set(self, key: str, value):
             raise RpcClientException(f'SCPreferencesSetValue failed to set: {key}')
         self._commit()
 
+    def set_dict(self, d: typing.Mapping):
+        for k, v in d.items():
+            self.set(k, v)
+
     def remove(self, key: str):
         if not self._client.symbols.SCPreferencesRemoveValue(self._ref, self._client.cf(key)):
             raise RpcClientException(f'SCPreferencesRemoveValue failed to remove: {key}')
@@ -33,8 +37,12 @@ def to_dict(self) -> typing.Mapping:
             result[k] = self.get(k)
         return result
 
+    def clear(self):
+        for k in self.keys:
+            self.remove(k)
+
     def _deallocate(self):
-        self._client.CFRelase(self._ref)
+        self._client.symbols.CFRelease(self._ref)
 
     def _commit(self):
         if not self._client.symbols.SCPreferencesCommitChanges(self._ref):

src/rpcclient/rpcclient/darwin/symbol.py

@@ -26,6 +26,9 @@ def cfdesc(self):
             return None
         return self._client.symbols.CFCopyDescription(self).py
 
+    def _decode_cfnull(self) -> None:
+        return None
+
     def _decode_cfstr(self) -> str:
         ptr = self._client.symbols.CFStringGetCStringPtr(self, 0)
         if ptr:
@@ -80,6 +83,7 @@ def py(self):
 
         t = self._client._cf_types[self._client.symbols.CFGetTypeID(self)]
         type_decoders = {
+            'null': self._decode_cfnull,
             'str': self._decode_cfstr,
             'bool': self._decode_cfbool,
             'number': self._decode_cfnumber,

src/rpcclient/tests/test_core_foundation_types.py

@@ -1,12 +1,16 @@
 import pytest
 
 
-@pytest.mark.parametrize('data', ['string',
-                                  b'bytes',
-                                  123,
-                                  0.1,
-                                  [0, 1, 'abc'],
-                                  {'key': 'value'},
-                                  [{'key': 'value'}, [1, 2]]])
+@pytest.mark.parametrize('data', [
+    None,
+    True,
+    False,
+    'string',
+    b'bytes',
+    123,
+    0.1,
+    [0, 1, 'abc'],
+    {'key': 'value'},
+    [{'key': 'value'}, [1, 2]]])
 def test_serialization(client, data):
     assert client.cf(data).py == data

src/rpcclient/tests/test_preferences.py

@@ -0,0 +1,44 @@
+DOMAIN = 'rpcserver'
+
+
+def test_cf_preferences(client):
+    if client.preferences.cf.get_keys(DOMAIN) is not None:
+        # if from some reason this domain already exist, empty it
+        client.preferences.cf.clear(DOMAIN)
+        assert client.preferences.cf.get_keys(DOMAIN) is None
+
+    # test set a full dictionary
+    test_dict = {'KEY1': 'VALUE1', 'KEY2': 'VALUE2'}
+    client.preferences.cf.set_dict(test_dict, DOMAIN)
+    assert test_dict == client.preferences.cf.get_values(DOMAIN)
+
+    # test set remove a single value
+    client.preferences.cf.remove('KEY2', DOMAIN)
+    test_dict.pop('KEY2')
+    assert test_dict == client.preferences.cf.get_values(DOMAIN)
+
+    # remove out test data and verify it works
+    client.preferences.cf.clear(DOMAIN)
+    assert client.preferences.cf.get_keys(DOMAIN) is None
+
+
+def test_sc_preferences(client):
+    with client.preferences.sc.get_preferences_object(DOMAIN) as pref:
+        if pref.to_dict() != {}:
+            # if from some reason this domain already exist, empty it
+            pref.clear()
+            assert pref.to_dict() == {}
+
+        # test set a full dictionary
+        test_dict = {'KEY1': 'VALUE1', 'KEY2': 'VALUE2'}
+        pref.set_dict(test_dict)
+        assert test_dict == pref.to_dict()
+
+        # test set remove a single value
+        pref.remove('KEY2')
+        test_dict.pop('KEY2')
+        assert test_dict == pref.to_dict()
+
+        # remove out test data and verify it works
+        pref.clear()
+        assert pref.to_dict() == {}