diff options
| -rw-r--r-- | libpathod/language/base.py | 29 | ||||
| -rw-r--r-- | libpathod/language/websockets.py | 69 | ||||
| -rw-r--r-- | libpathod/templates/docs_lang_websockets.html | 40 | ||||
| -rw-r--r-- | test/test_language_base.py | 21 | ||||
| -rw-r--r-- | test/test_language_websocket.py | 32 |
5 files changed, 177 insertions, 14 deletions
diff --git a/libpathod/language/base.py b/libpathod/language/base.py index 725a0d42..4179fa7d 100644 --- a/libpathod/language/base.py +++ b/libpathod/language/base.py @@ -224,7 +224,6 @@ class _Component(Token): A value component of the primary specification of an message. Components produce byte values desribe the bytes of the message. """ - @abc.abstractmethod def values(self, settings): # pragma: no cover """ A sequence of values, which can either be strings or generators. @@ -376,6 +375,34 @@ class Value(_Component): return self.__class__(self.value.freeze(settings)) +class Boolean(_Component): + """ + A boolean flag. + name = true + -name = false + """ + name = "" + + def __init__(self, value): + self.value = value + + @classmethod + def expr(klass): + e = pp.Optional(pp.Literal("-"), default=True) + e += pp.Literal(klass.name).suppress() + + def parse(s, loc, toks): + val = True + if toks[0] == "-": + val = False + return klass(val) + + return e.setParseAction(parse) + + def spec(self): + return "%s%s"%("-" if not self.value else "", self.name) + + class IntField(_Component): """ An integer field, where values can optionally specified by name. diff --git a/libpathod/language/websockets.py b/libpathod/language/websockets.py index ddffdab9..4b8a8148 100644 --- a/libpathod/language/websockets.py +++ b/libpathod/language/websockets.py @@ -17,7 +17,7 @@ class WF(base.CaselessLiteral): TOK = "wf" -class Code(base.IntField): +class OpCode(base.IntField): names = { "continue": netlib.websockets.OPCODE.CONTINUE, "text": netlib.websockets.OPCODE.TEXT, @@ -34,13 +34,46 @@ class Body(base.Value): preamble = "b" +class Raw(base.CaselessLiteral): + TOK = "r" + + +class Fin(base.Boolean): + name = "fin" + + +class RSV1(base.Boolean): + name = "rsv1" + + +class RSV2(base.Boolean): + name = "rsv2" + + +class RSV3(base.Boolean): + name = "rsv3" + + +class Mask(base.Boolean): + name = "mask" + + class WebsocketFrame(message.Message): comps = ( Body, - Code, + + OpCode, + # Bit flags + Fin, + RSV1, + RSV2, + RSV3, + Mask, actions.PauseAt, actions.DisconnectAt, - actions.InjectAt + actions.InjectAt, + + Raw, ) logattrs = ["body"] @property @@ -52,8 +85,28 @@ class WebsocketFrame(message.Message): return self.tok(Body) @property - def code(self): - return self.tok(Code) + def opcode(self): + return self.tok(OpCode) + + @property + def fin(self): + return self.tok(Fin) + + @property + def rsv1(self): + return self.tok(RSV1) + + @property + def rsv2(self): + return self.tok(RSV2) + + @property + def rsv3(self): + return self.tok(RSV3) + + @property + def mask(self): + return self.tok(Mask) @classmethod def expr(klass): @@ -81,8 +134,10 @@ class WebsocketFrame(message.Message): mask = True, payload_length = length ) - if self.code: - frameparts["opcode"] = self.code.value + for i in ["opcode", "fin", "rsv1", "rsv2", "rsv3", "mask"]: + v = getattr(self, i, None) + if v is not None: + frameparts[i] = v.value frame = netlib.websockets.FrameHeader(**frameparts) vals = [frame.to_bytes()] if self.body: diff --git a/libpathod/templates/docs_lang_websockets.html b/libpathod/templates/docs_lang_websockets.html index c414320c..9b595e74 100644 --- a/libpathod/templates/docs_lang_websockets.html +++ b/libpathod/templates/docs_lang_websockets.html @@ -16,9 +16,45 @@ <td> Set the op code. This can either be an integer from 0-15, or be - one of the following opcode names: <b>continue</b>, - <b>text</b>, <b>binary</b>, <b>close</b>, <b>ping</b>, + one of the following opcode names: <b>text</b> (the default), + <b>continue</b>, <b>binary</b>, <b>close</b>, <b>ping</b>, <b>pong</b>. + + </td> + </tr> + + <tr> + <td> [-]fin </td> + <td> + Set or un-set the <b>fin</b> bit. + </td> + </tr> + + <tr> + <td> [-]rsv1 </td> + <td> + Set or un-set the <b>rsv1</b> bit. + </td> + </tr> + + <tr> + <td> [-]rsv2 </td> + <td> + Set or un-set the <b>rsv2</b> bit. + </td> + </tr> + + <tr> + <td> [-]rsv3 </td> + <td> + Set or un-set the <b>rsv3</b> bit. + </td> + </tr> + + <tr> + <td> [-]mask </td> + <td> + Set or un-set the <b>mask</b> bit. </td> </tr> diff --git a/test/test_language_base.py b/test/test_language_base.py index 4149de3e..c6cee7b3 100644 --- a/test/test_language_base.py +++ b/test/test_language_base.py @@ -280,3 +280,24 @@ def test_integer(): assert v.spec() == "200" assert v.freeze({}).value == v.value + + +class TBoolean(base.Boolean): + name = "test" + + +class test_boolean(): + e = TBoolean.expr() + assert e.parseString("test")[0].value + assert not e.parseString("-test")[0].value + + def roundtrip(s): + e = TBoolean.expr() + s2 = e.parseString(s)[0].spec() + v1 = e.parseString(s)[0].value + v2 = e.parseString(s2)[0].value + assert s == s2 + assert v1 == v2 + + roundtrip("test") + roundtrip("-test") diff --git a/test/test_language_websocket.py b/test/test_language_websocket.py index f2f0b2a8..e313c591 100644 --- a/test/test_language_websocket.py +++ b/test/test_language_websocket.py @@ -15,7 +15,11 @@ class TestWebsocketFrame: "wf", "wf:b'foo'", "wf:cbinary", - "wf:c1" + "wf:c1", + "wf:r", + "wf:fin", + "wf:fin:rsv1:rsv2:rsv3:mask", + "wf:-fin:-rsv1:-rsv2:-rsv3:-mask", ] for i in specs: wf = parse_request(i) @@ -28,12 +32,32 @@ class TestWebsocketFrame: wf2 = parse_request(spec) assert wf2.spec() == spec + def test_raw(self): + pass + + def test_flags(self): + wf = parse_request("wf:fin:mask:rsv1:rsv2:rsv3") + frm = netlib.websockets.Frame.from_bytes(tutils.render(wf)) + assert frm.header.fin + assert frm.header.mask + assert frm.header.rsv1 + assert frm.header.rsv2 + assert frm.header.rsv3 + + wf = parse_request("wf:-fin:-mask:-rsv1:-rsv2:-rsv3") + frm = netlib.websockets.Frame.from_bytes(tutils.render(wf)) + assert not frm.header.fin + assert not frm.header.mask + assert not frm.header.rsv1 + assert not frm.header.rsv2 + assert not frm.header.rsv3 + def test_construction(self): wf = parse_request("wf:c1") frm = netlib.websockets.Frame.from_bytes(tutils.render(wf)) - assert wf.code.value == 1 == frm.header.opcode + assert wf.opcode.value == 1 == frm.header.opcode wf = parse_request("wf:cbinary") frm = netlib.websockets.Frame.from_bytes(tutils.render(wf)) - assert wf.code.value == frm.header.opcode - assert wf.code.value == netlib.websockets.OPCODE.BINARY + assert wf.opcode.value == frm.header.opcode + assert wf.opcode.value == netlib.websockets.OPCODE.BINARY |