1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
#!/usr/bin/env python
"""Test custom JSON encoding by subclassing VarlinkEncoder and
passing it to a server Service.
"""
import dataclasses
import os
import threading
import time
import typing
import unittest
import varlink
import varlink.error
@dataclasses.dataclass
class Shipment:
name: str
description: str
size: int
weight: typing.Optional[int] = None
@dataclasses.dataclass
class Order:
shipments: list[Shipment]
order_num: int
customer: str
@dataclasses.dataclass
class GetOrderResult:
order: Order
class VDCEncoder(varlink.error.VarlinkEncoder):
def default(self, obj):
if dataclasses.is_dataclass(obj):
return dataclasses.asdict(obj)
return super().default(obj)
service = varlink.Service(
vendor="Varlink",
product="Varlink Encoding Example",
version="1",
url="http://varlink.org",
interface_dir=os.path.dirname(__file__),
json_encoder_cls=VDCEncoder,
)
class ServiceRequestHandler(varlink.RequestHandler):
service = service
@service.interface("org.example.encoding")
class EncodingExample:
sleep_duration = 1
def Ping(self, ping):
return {"pong": ping}
def GetOrder(self, num):
order = Order(
shipments=[
Shipment(
name="Furniture",
description="Furniture by Ferb",
size=1000,
weight=400,
),
Shipment(
name="Electronics",
description="Electronics by Earl",
size=588,
),
],
order_num=num,
customer="Joe's Discount Store",
)
return GetOrderResult(order=order)
class TestService(unittest.TestCase):
address = "tcp:127.0.0.1:23451"
@classmethod
def setUpClass(cls):
cls._server = varlink.ThreadingServer(cls.address, ServiceRequestHandler)
cls._server_thread = threading.Thread(target=cls._server.serve_forever)
cls._server_thread.start()
time.sleep(0.1)
@classmethod
def tearDownClass(cls):
cls._server.shutdown()
cls._server.server_close()
cls._server_thread.join()
def test_ping(self):
client = varlink.Client.new_with_address(self.address)
with client.open("org.example.encoding") as conn:
response = conn.Ping("Foo")
self.assertEqual(response["pong"], "Foo")
def test_get_order(self):
client = varlink.Client.new_with_address(self.address)
with client.open("org.example.encoding") as conn:
response = conn.GetOrder(4638547)
# response will be a dict represenation of GetOrderResult
self.assertIn("order", response)
order = response["order"]
self.assertEqual(order.get("order_num"), 4638547)
self.assertEqual(order.get("customer"), "Joe's Discount Store")
self.assertEqual(len(order.get("shipments", [])), 2)
shipment1 = order["shipments"][0]
self.assertEqual(shipment1.get("name"), "Furniture")
self.assertIsNotNone(shipment1.get("weight"))
shipment2 = order["shipments"][1]
self.assertEqual(shipment2.get("name"), "Electronics")
self.assertIsNone(shipment2.get("weight"))
|
