cert

varlink-certification/org.varlink.certification.varlink

@@ -0,0 +1,66 @@
+interface org.varlink.certification
+
+type MyOtherType (
+  foo: ?[]?[string](foo, bar, baz),
+  anon: (foo: bool, bar: bool)
+)
+
+type MyType (
+  object: object,
+  enum: (one, two, three),
+  struct: (first: int, second: string),
+  array: []string,
+  dictionary: [string]string,
+  stringset: [string](),
+  nullable: ?string,
+  nullable_array_struct: ?[](first: int, second: string),
+  other_type: MyOtherType
+)
+
+method Start() -> ()
+
+method Test01() -> (bool: bool)
+
+method Test02(bool: bool) -> (int: int)
+
+method Test03(int: int) -> (float: float)
+
+method Test04(float: float) -> (string: string)
+
+method Test05(string: string) -> (
+  bool: bool,
+  int: int,
+  float: float,
+  string: string
+)
+
+method Test06(
+  bool: bool,
+  int: int,
+  float: float,
+  string: string
+) -> (
+  struct: (
+    bool: bool,
+    int: int,
+    float: float,
+    string: string
+  )
+)
+
+method Test07(
+  struct: (
+    bool: bool,
+    int: int,
+    float: float,
+    string: string
+  )
+) -> (map: [string]string)
+
+method Test08(map: [string]string) -> (set: [string]())
+
+method Test09(set: [string]()) -> (mytype: MyType)
+
+method End(mytype: MyType) -> ()
+
+error CertificationError (wants: string, got: string)

varlink-certification/varlink-certification-server.py

@@ -0,0 +1,207 @@
+import varlink
+import sys
+import os
+import json
+
+service = varlink.Service(
+    vendor='Varlink',
+    product='Varlink Examples',
+    version='1',
+    url='http://varlink.org',
+    interface_dir=os.path.dirname(__file__)
+)
+
+
+def sorted_json(dct):
+    if isinstance(dct, type([])):
+        return sorted(dct)
+    return dct
+
+
+def handle_message(message):
+    print("IN :", message)
+    ret = service.handle(message)
+    for out in ret:
+        print("OUT:", out)
+        yield out
+
+
+class CertificationError(varlink.VarlinkError):
+
+    def __init__(self, wants, got):
+        varlink.VarlinkError.__init__(self,
+                                      {'error': 'org.varlink.certification.CertificationError',
+                                       'parameters': {'wants': wants, 'got': got}})
+
+
+@service.interface('org.varlink.certification')
+class CertService:
+    def assert_raw(self, _raw, _message, wants):
+
+        have = json.loads(wants, object_hook=sorted_json)
+
+        if have != _message:
+            print("Got :", have)
+            print("Want:", _message)
+            raise CertificationError(wants, _raw.decode("utf-8"))
+
+    def assert_cmp(self, _raw, wants, _bool):
+        if not _bool:
+            raise CertificationError(wants, _raw.decode("utf-8"))
+
+    def Start(self, _raw=None, _message=None):
+        print("Start:", _raw, _message)
+        self.assert_raw(_raw, _message, '{"method": "org.varlink.certification.Start", "parameters": {}}')
+
+    # () -> (bool: bool)
+    def Test01(self, _raw=None, _message=None):
+        self.assert_raw(_raw, _message, '{"method": "org.varlink.certification.Test01", "parameters": {}}')
+        return {"bool": True}
+
+    # (bool: bool) -> (int: int)
+    def Test02(self, _bool, _raw=None, _message=None):
+        wants = '{"method": "org.varlink.certification.Test02", "parameters": {"bool": true}}'
+        self.assert_cmp(_raw, wants, _bool == True)
+        self.assert_raw(_raw, _message, wants)
+        return {"int": 1}
+
+    # (int: int) -> (float: float)
+    def Test03(self, _int, _raw=None, _message=None):
+        wants = '{"method": "org.varlink.certification.Test03", "parameters": {"int": 1}}'
+        self.assert_cmp(_raw, wants, _int == 1)
+        self.assert_raw(_raw, _message, wants)
+        return {"float": 1.0}
+
+    # (float: float) -> (string: string)
+    def Test04(self, _float, _raw=None, _message=None):
+        wants = '{"method": "org.varlink.certification.Test04", "parameters": {"float": 1.0}}'
+        self.assert_cmp(_raw, wants, _float == 1.0)
+        self.assert_raw(_raw, _message, wants)
+        return {"string": "ping"}
+
+    # (string: string) -> (bool: bool, int: int, float: float, string: string)
+    def Test05(self, _string, _raw=None, _message=None):
+        wants = '{"method": "org.varlink.certification.Test05", "parameters": {"string": "ping"}}'
+        self.assert_cmp(_raw, wants, _string == "ping")
+        self.assert_raw(_raw, _message, wants)
+        return {"bool": False, "int": 2, "float": 3.1415, "string": "a lot of string"}
+
+    # (bool: bool, int: int, float: float, string: string)
+    # -> (struct: (bool: bool, int: int, float: float, string: string))
+    def Test06(self, _bool, _int, _float, _string, _raw=None, _message=None):
+        wants = '{"method": "org.varlink.certification.Test06", "parameters": ' \
+                '{"bool": false, "int": 2, "float": 3.1415, "string": "a lot of ' \
+                'string"}}'
+        self.assert_raw(_raw, _message, wants)
+        self.assert_cmp(_raw, wants, _int == 2)
+        self.assert_cmp(_raw, wants, _bool == False)
+        self.assert_cmp(_raw, wants, _float == 3.1415)
+        self.assert_cmp(_raw, wants, _string == "a lot of string")
+
+        return {"struct": {"bool": False, "int": 2, "float": 3.1415, "string": "a lot of string"}}
+
+    # (struct: (bool: bool, int: int, float: float, string: string)) -> (map: [string]string)
+    def Test07(self, _dict, _raw=None, _message=None):
+        wants = '{"method": "org.varlink.certification.Test07", "parameters": ' \
+                '{"struct": {"int": 2, "bool": false, "float": 3.1415, "string": "a lot of ' \
+                'string"}}}'
+        self.assert_raw(_raw, _message, wants)
+        self.assert_cmp(_raw, wants, _dict["int"] == 2)
+        self.assert_cmp(_raw, wants, _dict["bool"] == False)
+        self.assert_cmp(_raw, wants, _dict["float"] == 3.1415)
+        self.assert_cmp(_raw, wants, _dict["string"] == "a lot of string")
+        return {"map": {"foo": "Foo", "bar": "Bar"}}
+
+    # (map: [string]string) -> (set: [string]())
+    def Test08(self, _map, _raw=None, _message=None):
+        print(map, file=sys.stderr)
+        self.assert_raw(_raw, _message,
+                        '{"method": "org.varlink.certification.Test08", "parameters": {"map" : {"foo": "Foo", '
+                        '"bar": "Bar"}}}')
+        return {"set": {"one", "two", "three"}}
+
+    # (set: [string]()) -> (mytype: MyType)
+    def Test09(self, _set, _raw=None, _message=None):
+        wants = '{"method": "org.varlink.certification.Test09", "parameters": {"set" : {"one": null, "three": null,' \
+                ' "two": null}}}'
+        print(_set, file=sys.stderr)
+        self.assert_raw(_raw, _message, wants)
+        self.assert_cmp(_raw, wants, isinstance(_set, set))
+        self.assert_cmp(_raw, wants, len(_set) == 3)
+        self.assert_cmp(_raw, wants, "one" in _set)
+        self.assert_cmp(_raw, wants, "two" in _set)
+        self.assert_cmp(_raw, wants, "three" in _set)
+        return {"mytype": {
+            "object": {"method": "org.varlink.certification.Test09",
+                       "parameters": {"map": {"foo": "Foo", "bar": "Bar"}}},
+            "enum": "two",
+            "struct": {"first": 1, "second": "2"},
+            "array": ["one", "two", "three"],
+            "dictionary": {"foo": "Foo", "bar": "Bar"},
+            "stringset": {"one", "two", "three"},
+            "nullable": None,
+            "nullable_array_struct": None,
+            "other_type": {
+                "foo": [
+                    None,
+                    {"foo": "foo", "bar": "bar"},
+                    None,
+                    {"one": "foo", "two": "bar"}
+                ],
+                "anon": {"foo": True, "bar": False}
+            }
+        }}
+
+    # (mytype: MyType) -> ()
+    def End(self, mytype, _raw=None, _message=None):
+        wants = '{"method": "org.varlink.certification.End", "parameters": {"mytype": {"object": {"method": ' \
+                '"org.varlink.certification.Test09", "parameters": ' \
+                '{"map": {"foo": "Foo", "bar": "Bar"}}}, "enum": "two", "struct": {"first": 1, "second": "2"}, ' \
+                '"array": ["one", "two", "three"], "dictionary": {"foo": "Foo", "bar": "Bar"}, "stringset": ' \
+                '{"two": null, "one": null, "three": null}, ' \
+                '"other_type": {"foo": [null, {"foo": "foo", "bar": "bar"}, null, {"one": "foo", "two": "bar"}], ' \
+                '"anon": {"foo": true, "bar": false}}}}}'
+        print(mytype, file=sys.stderr)
+
+        if "nullable" in mytype:
+            self.assert_cmp(_raw, wants, mytype["nullable"] == None)
+            del mytype["nullable"]
+
+        if "nullable_array_struct" in mytype:
+            self.assert_cmp(_raw, wants, mytype["nullable_array_struct"] == None)
+            del mytype["nullable_array_struct"]
+
+        self.assert_cmp(_raw, wants, mytype == {
+            "object": {"method": "org.varlink.certification.Test09",
+                       "parameters": {"map": {"foo": "Foo", "bar": "Bar"}}},
+            "enum": "two",
+            "struct": {"first": 1, "second": "2"},
+            "array": ["one", "two", "three"],
+            "dictionary": {"foo": "Foo", "bar": "Bar"},
+            "stringset": {"one", "two", "three"},
+            "other_type": {
+                "foo": [
+                    None,
+                    {"foo": "foo", "bar": "bar"},
+                    None,
+                    {"one": "foo", "two": "bar"}
+                ],
+                "anon": {"foo": True, "bar": False}
+            }
+        })
+        self.assert_raw(_raw, _message, wants)
+
+
+if __name__ == '__main__':
+    if len(sys.argv) < 2 or not sys.argv[1].startswith("--varlink="):
+        print('Usage: %s --varlink=<varlink address>' % sys.argv[0])
+        sys.exit(1)
+
+    with varlink.SimpleServer(service) as s:
+        print("Listening on", sys.argv[1][10:])
+        try:
+            sock = s.get_socket(sys.argv[1][10:])
+            s.listen(sock, handle_message)
+        except KeyboardInterrupt:
+            pass
+    sys.exit(0)