diff --git a/src/rpcclient/rpcclient/client.py b/src/rpcclient/rpcclient/client.py index 61643a67..df5e7c4e 100644 --- a/src/rpcclient/rpcclient/client.py +++ b/src/rpcclient/rpcclient/client.py @@ -263,6 +263,10 @@ def environ(self) -> typing.List[str]: i += 1 return result + @property + def pid(self): + return int(self.symbols.getpid()) + @contextlib.contextmanager def safe_calloc(self, size: int): with self.safe_malloc(size) as x: diff --git a/src/rpcclient/tests/test_allocation_cleanup.py b/src/rpcclient/tests/test_allocation_cleanup.py new file mode 100644 index 00000000..63a7abca --- /dev/null +++ b/src/rpcclient/tests/test_allocation_cleanup.py @@ -0,0 +1,43 @@ +import gc + + +def test_allocate_file_fd_context_manager(client, tmp_path): + # make sure when the test starts, all previous Allocated references are freed + gc.collect() + fds_count = len(client.processes.get_fds(client.pid)) + with client.fs.open(tmp_path / 'test', 'w'): + assert fds_count + 1 == len(client.processes.get_fds(client.pid)) + assert fds_count == len(client.processes.get_fds(client.pid)) + + +def test_allocate_file_fd_gc(client, tmp_path): + # make sure when the test starts, all previous Allocated references are freed + gc.collect() + fds_count = len(client.processes.get_fds(client.pid)) + + # create a new fd with zero references, so it should be free immediately + client.fs.open(tmp_path / 'test', 'w') + + # make sure python's GC had a chance to free the newly created fd + gc.collect() + assert fds_count == len(client.processes.get_fds(client.pid)) + + +def test_allocate_file_fd_explicit_del(client, tmp_path): + # make sure when the test starts, all previous Allocated references are freed + gc.collect() + fds_count = len(client.processes.get_fds(client.pid)) + fd = client.fs.open(tmp_path / 'test', 'w') + assert fds_count + 1 == len(client.processes.get_fds(client.pid)) + del fd + assert fds_count == len(client.processes.get_fds(client.pid)) + + +def test_allocate_file_fd_explicit_deallocate(client, tmp_path): + # make sure when the test starts, all previous Allocated references are freed + gc.collect() + fds_count = len(client.processes.get_fds(client.pid)) + fd = client.fs.open(tmp_path / 'test', 'w') + assert fds_count + 1 == len(client.processes.get_fds(client.pid)) + fd.deallocate() + assert fds_count >= len(client.processes.get_fds(client.pid)) diff --git a/src/rpcclient/tests/test_core_foundation_types.py b/src/rpcclient/tests/test_core_foundation_types.py new file mode 100644 index 00000000..3dbc835f --- /dev/null +++ b/src/rpcclient/tests/test_core_foundation_types.py @@ -0,0 +1,12 @@ +import pytest + + +@pytest.mark.parametrize('data', ['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