librclone: add basic Python bindings with tests #4891

This commit is contained in:
Nick Craig-Wood 2021-04-28 16:53:35 +01:00
parent 665eceaec3
commit e33303df94
4 changed files with 144 additions and 1 deletions

View File

@ -199,6 +199,7 @@ jobs:
run: |
make -C librclone/ctest test
make -C librclone/ctest clean
librclone/python/test_rclone.py
if: matrix.librclonetest
- name: Code quality test

View File

@ -38,9 +38,15 @@ There is an example program `ctest.c` with Makefile in the `ctest` subdirectory
## gomobile
The gomobile subdirectory contains the equivalent of the C binding but
The `gomobile` subdirectory contains the equivalent of the C binding but
suitable for using with gomobile using something like this.
gomobile bind -v -target=android github.com/rclone/rclone/librclone/gomobile
## python
The `python` subdirectory contains a simple python wrapper for the C
API using rclone linked as a shared library.
This needs expanding and submitting to pypi...

View File

@ -0,0 +1,94 @@
"""
Python interface to librclone.so using ctypes
Create an rclone object
rclone = Rclone(shared_object="/path/to/librclone.so")
Then call rpc calls on it
rclone.rpc("rc/noop", a=42, b="string", c=[1234])
When finished, close it
rclone.close()
"""
__all__ = ('Rclone', 'RcloneException')
import os
import json
import subprocess
from ctypes import *
class RcloneRPCResult(Structure):
"""
This is returned from the C API when calling RcloneRPC
"""
_fields_ = [("Output", c_char_p),
("Status", c_int)]
class RcloneException(Exception):
"""
Exception raised from rclone
This will have the attributes:
output - a dictionary from the call
status - a status number
"""
def __init__(self, output, status):
self.output = output
self.status = status
message = self.output.get('error', 'Unknown rclone error')
super().__init__(message)
class Rclone():
"""
Interface to Rclone via librclone.so
Initialise with shared_object as the file path of librclone.so
"""
def __init__(self, shared_object="./librclone.so"):
self.rclone = CDLL(shared_object)
self.rclone.RcloneRPC.restype = RcloneRPCResult
self.rclone.RcloneRPC.argtypes = (c_char_p, c_char_p)
self.rclone.RcloneInitialize.restype = None
self.rclone.RcloneInitialize.argtypes = ()
self.rclone.RcloneFinalize.restype = None
self.rclone.RcloneFinalize.argtypes = ()
self.rclone.RcloneInitialize()
def rpc(self, method, **kwargs):
"""
Call an rclone RC API call with the kwargs given.
The result will be a dictionary.
If an exception is raised from rclone it will of type
RcloneException.
"""
method = method.encode("utf-8")
parameters = json.dumps(kwargs).encode("utf-8")
resp = self.rclone.RcloneRPC(method, parameters)
output = json.loads(resp.Output.decode("utf-8"))
status = resp.Status
if status != 200:
raise RcloneException(output, status)
return output
def close(self):
"""
Call to finish with the rclone connection
"""
self.rclone.RcloneFinalize()
self.rclone = None
@classmethod
def build(cls, shared_object):
"""
Builds rclone to shared_object if it doesn't already exist
Requires go to be installed
"""
if os.path.exists(shared_object):
return
print("Building "+shared_object)
subprocess.check_call(["go", "build", "--buildmode=c-shared", "-o", shared_object, "github.com/rclone/rclone/librclone"])

42
librclone/python/test_rclone.py Executable file
View File

@ -0,0 +1,42 @@
#!/usr/bin/env python3
"""
Test program for librclone
"""
import os
import subprocess
import unittest
from rclone import *
class TestRclone(unittest.TestCase):
"""TestSuite for rclone python module"""
shared_object = "librclone.so"
@classmethod
def setUpClass(cls):
super(TestRclone, cls).setUpClass()
cls.shared_object = "./librclone.so"
Rclone.build(cls.shared_object)
cls.rclone = Rclone(shared_object=cls.shared_object)
@classmethod
def tearDownClass(cls):
cls.rclone.close()
os.remove(cls.shared_object)
super(TestRclone, cls).tearDownClass()
def test_rpc(self):
o = self.rclone.rpc("rc/noop", a=42, b="string", c=[1234])
self.assertEqual(dict(a=42, b="string", c=[1234]), o)
def test_rpc_error(self):
try:
o = self.rclone.rpc("rc/error", a=42, b="string", c=[1234])
except RcloneException as e:
self.assertEqual(e.status, 500)
self.assertTrue(e.output["error"].startswith("arbitrary error"))
else:
raise ValueError("Expecting exception")
if __name__ == '__main__':
unittest.main()