mirror of
https://gitlab.com/buildroot.org/buildroot.git
synced 2026-09-30 22:11:59 -09:00
support/testing: python-varlink: new runtime test
Add new runtime test for python-varlink. As always when we need to start a separate server process inside a test case this gets slightly fiddly. We override the test_run() function to first start the varlink example server, then call the packages cli interface to do a varlink call against the server. The cli defaults to pretty printing the result, which makes it more annoying to compare to the expected result in the test case, so we un-prettyprint it with python's builtin json.tool module. Signed-off-by: Marcus Hoffmann <buildroot@bubu1.eu> Signed-off-by: Thomas Petazzoni <thomas.petazzoni@bootlin.com>
This commit is contained in:
committed by
Thomas Petazzoni
parent
ec0872b0e0
commit
33479613d4
30
support/testing/tests/package/test_python_varlink.py
Normal file
30
support/testing/tests/package/test_python_varlink.py
Normal file
@@ -0,0 +1,30 @@
|
||||
import os
|
||||
|
||||
from tests.package.test_python import TestPythonPackageBase
|
||||
|
||||
|
||||
class TestPythonPy3Varlink(TestPythonPackageBase):
|
||||
__test__ = True
|
||||
config = TestPythonPackageBase.config + \
|
||||
"""
|
||||
BR2_PACKAGE_PYTHON3=y
|
||||
BR2_PACKAGE_PYTHON_VARLINK=y
|
||||
"""
|
||||
sample_scripts = ["tests/package/test_python_varlink/sample_python_varlink.py",
|
||||
"tests/package/test_python_varlink/org.example.test.varlink"]
|
||||
|
||||
def test_run(self):
|
||||
self.login()
|
||||
self.check_sample_scripts_exist()
|
||||
# we need to supress any output here as otherwise the commnd output parsing
|
||||
# gets confused for the second command
|
||||
cmd = "%s %s > /dev/null 2>&1 &" % (self.interpreter, os.path.basename(self.sample_scripts[0]))
|
||||
|
||||
_, exit_code = self.emulator.run(cmd, timeout=self.timeout)
|
||||
|
||||
varlink_cli_call = "-m varlink.cli call 'unix:@test/org.example.test.Ping' '{\"ping\": \"hello\"}'"
|
||||
unprettyprint = "| python -m json.tool --compact"
|
||||
cmd = "%s %s %s" % (self.interpreter, varlink_cli_call, unprettyprint)
|
||||
output, exit_code = self.emulator.run(cmd, timeout=self.timeout)
|
||||
self.assertEqual(exit_code, 0)
|
||||
self.assertEqual(output[0], '{"pong":"hello"}')
|
||||
@@ -0,0 +1,5 @@
|
||||
# Example Varlink service
|
||||
interface org.example.test
|
||||
|
||||
# Returns the same string
|
||||
method Ping(ping: string) -> (pong: string)
|
||||
@@ -0,0 +1,24 @@
|
||||
import os
|
||||
|
||||
import varlink
|
||||
|
||||
service = varlink.Service(interface_dir=os.path.dirname(__file__))
|
||||
|
||||
|
||||
class ServiceRequestHandler(varlink.RequestHandler):
|
||||
service = service
|
||||
|
||||
|
||||
@service.interface("org.example.test")
|
||||
class Example:
|
||||
def Ping(self, ping):
|
||||
return {"pong": ping}
|
||||
|
||||
|
||||
def run_server(address):
|
||||
with varlink.ThreadingServer(address, ServiceRequestHandler) as server:
|
||||
print("Listening on", server.server_address)
|
||||
server.serve_forever()
|
||||
|
||||
|
||||
run_server("unix:@test")
|
||||
Reference in New Issue
Block a user