Merge pull request #2410 from bandprotocol/add-feature-pyobi-and-test…

…case

pyobi: improve code structure, add PyObiBool and add more tests

CHANGELOG_UNRELEASED.md

@@ -49,6 +49,8 @@
 
 ### Oracle Binary Encoding (OBI)
 
+- (impv) [\#2410](https://github.com/bandprotocol/bandchain/pull/2410) Improve code structure, Add PyObiBool and add more tests
+
 ### Helpers
 
 ### MISC

obi/pyobi/README.md

@@ -0,0 +1,15 @@
+# PyObi
+
+### setup
+
+```
+python3 -m venv venv
+source venv/bin/activate
+pip install -r requirements.txt
+```
+
+### testing
+
+```
+PYTHONPATH=. pytest
+```

obi/pyobi/pyobi/__init__.py

@@ -2,6 +2,7 @@
     PyObi,
     PyObiSpec,
     PyObiInteger,
+    PyObiBool,
     PyObiVector,
     PyObiStruct,
     PyObiString,

obi/pyobi/pyobi/pyobi.py

@@ -1,6 +1,3 @@
-import re
-
-
 class PyObiSpec(object):
     impls = []
 
@@ -11,13 +8,17 @@ def __init_subclass__(cls, **kwargs):
     @classmethod
     def from_spec(cls, spec):
         for impl in cls.impls:
-            if re.match(impl.REGEX, spec):
+            if impl.match_schema(spec):
                 return impl(spec)
         raise ValueError("Cannot parse spec: {}".format(spec))
 
     def __init__(self, spec):
         raise NotImplementedError()
 
+    @classmethod
+    def match_schema(cls, schema):
+        raise NotImplementedError()
+
     def encode(self, value):
         raise NotImplementedError()
 
@@ -26,28 +27,52 @@ def decode(self, data):
 
 
 class PyObiInteger(PyObiSpec):
-    REGEX = re.compile(r"^(u|i)(8|16|32|64|128|256)$")
-
     def __init__(self, spec):
         self.is_signed = spec[0] == "i"
         self.size_in_bytes = int(spec[1:]) // 8
 
+    @classmethod
+    def match_schema(cls, schema):
+        return schema[:1] in ["i", "u"] and schema[1:] in ["8", "16", "32", "64", "128", "256"]
+
     def encode(self, value):
         return value.to_bytes(self.size_in_bytes, byteorder="big", signed=self.is_signed)
 
     def decode(self, data):
         return (
-            int.from_bytes(data[: self.size_in_bytes], byteorder="big", signed=self.is_signed,),
+            int.from_bytes(data[: self.size_in_bytes], byteorder="big", signed=self.is_signed),
             data[self.size_in_bytes :],
         )
 
 
-class PyObiVector(PyObiSpec):
-    REGEX = re.compile(r"^\[.*\]$")
+class PyObiBool(PyObiSpec):
+    def __init__(self, spec="bool"):
+        pass
 
+    @classmethod
+    def match_schema(cls, schema):
+        return schema == "bool"
+
+    def encode(self, value):
+        return PyObiInteger("u8").encode(1 if value else 0)
+
+    def decode(self, data):
+        u8, remaining = PyObiInteger("u8").decode(data)
+        if u8 == 1:
+            return True, remaining
+        elif u8 == 0:
+            return False, remaining
+        raise ValueError("Boolean value must be 1 or 0 but got {}".format(u8))
+
+
+class PyObiVector(PyObiSpec):
     def __init__(self, spec):
         self.intl_obi = self.from_spec(spec[1:-1])
 
+    @classmethod
+    def match_schema(cls, schema):
+        return schema[0] == "[" and schema[-1] == "]"
+
     def encode(self, value):
         result = PyObiInteger("u32").encode(len(value))
         for each in value:
@@ -64,8 +89,6 @@ def decode(self, data):
 
 
 class PyObiStruct(PyObiSpec):
-    REGEX = re.compile(r"^{.*}$")
-
     def __init__(self, spec):
         self.intl_obi_kvs = []
         fields = ['']
@@ -85,6 +108,10 @@ def __init__(self, spec):
                 raise ValueError("Expect at least one colon for each struct field")
             self.intl_obi_kvs.append((tokens[0], self.from_spec(tokens[1])))
 
+    @classmethod
+    def match_schema(cls, schema):
+        return schema[0] == "{" and schema[-1] == "}"
+
     def encode(self, value):
         result = b""
         for key, spec in self.intl_obi_kvs:
@@ -99,11 +126,13 @@ def decode(self, data):
 
 
 class PyObiString(PyObiSpec):
-    REGEX = re.compile(r"^string$")
-
-    def __init__(self, spec):
+    def __init__(self, spec="string"):
         pass
 
+    @classmethod
+    def match_schema(cls, schema):
+        return schema == "string"
+
     def encode(self, value):
         return PyObiInteger("u32").encode(len(value)) + value.encode()
 
@@ -113,11 +142,13 @@ def decode(self, data):
 
 
 class PyObiBytes(PyObiSpec):
-    REGEX = re.compile(r"^bytes$")
-
-    def __init__(self, spec):
+    def __init__(self, spec="bytes"):
         pass
 
+    @classmethod
+    def match_schema(cls, schema):
+        return schema == "bytes"
+
     def encode(self, value):
         return PyObiInteger("u32").encode(len(value)) + value
 
@@ -128,27 +159,27 @@ def decode(self, data):
 
 class PyObi(object):
     def __init__(self, schema):
-        normalized_schema = re.sub(r"\s+", "", schema)
+        normalized_schema = "".join(schema.split())
         tokens = normalized_schema.split("/")
-        if len(tokens) != 2:
-            raise ValueError("Expect one forward slash in OBI schema")
-        self.input_schema = PyObiSpec.from_spec(tokens[0])
-        self.output_schema = PyObiSpec.from_spec(tokens[1])
+        self.schemas = [PyObiSpec.from_spec(token) for token in tokens]
 
-    def encode_input(self, value):
-        return self.input_schema.encode(value)
+    def encode(self, data, index=0):
+        return self.schemas[index].encode(data)
 
-    def decode_input(self, data):
-        result, remaining = self.input_schema.decode(data)
+    def decode(self, data, index=0):
+        result, remaining = self.schemas[index].decode(data)
         if remaining:
             raise ValueError("Not all data is consumed after decoding input")
         return result
 
-    def encode_output(self, value):
-        return self.output_schema.encode(value)
+    def encode_input(self, data):
+        return self.encode(data, index=0)
+
+    def encode_output(self, data):
+        return self.encode(data, index=1)
+
+    def decode_input(self, data):
+        return self.decode(data, index=0)
 
     def decode_output(self, data):
-        result, remaining = self.output_schema.decode(data)
-        if remaining:
-            raise ValueError("Not all data is consumed after decoding output")
-        return result
+        return self.decode(data, index=1)

obi/pyobi/requirements.txt

@@ -0,0 +1 @@
+pytest==6.0.1