Add some tests

.gitignore

@@ -1,5 +1,6 @@
 __pycache__/
 __pypackages__/
+.pytest_cache/
 *.pyc
 /build/
 /*.egg-info/

computercraft/server.py

@@ -5,8 +5,7 @@ from os.path import join, dirname, abspath
 
 from aiohttp import web, WSMsgType
 
-from .sess import CCSession
-from . import ser
+from . import ser, sess
 from .rproc import lua_table_to_list
 
 
@@ -14,7 +13,7 @@ THIS_DIR = dirname(abspath(__file__))
 LUA_FILE = join(THIS_DIR, 'back.lua')
 PROTO_VERSION = 4
 PROTO_ERROR = b'C' + ser.serialize(b'protocol error')
-DEBUG_PROTO = False
+DEBUG_PROTO = True
 
 
 async def _bin_messages(ws):
@@ -32,7 +31,7 @@ async def _send(ws, data):
     await ws.send_bytes(data)
 
 
-def protocol(send, sess_cls=CCSession):
+def protocol(send, sess_cls=sess.CCSession):
     # handle first frame
     msg = yield
     msg = ser.dcmditer(msg)
@@ -77,7 +76,7 @@ class CCApplication(web.Application):
         await ws.prepare(request)
 
         squeue = []
-        pgen = protocol(squeue.append)
+        pgen = self['protocol_factory'](squeue.append)
         next(pgen)
         mustquit = False
         async for msg in _bin_messages(ws):
@@ -107,7 +106,7 @@ class CCApplication(web.Application):
                 print('SEND', m)
             squeue.append(m)
 
-        pgen = protocol(send)
+        pgen = self['protocol_factory'](send)
         # pgen = protocol(squeue.append)
         next(pgen)
         mustquit = False
@@ -165,7 +164,7 @@ class CCApplication(web.Application):
         self.router.add_get('/ws/', self.ws)
 
 
-def main():
+def create_parser():
     parser = argparse.ArgumentParser()
     parser.add_argument('--host', default='0.0.0.0')
     parser.add_argument(
@@ -174,11 +173,18 @@ def main():
     parser.add_argument(
         '--oc-port', type=int, default=8001,
         help='Raw TCP port for opencomputers')
-    args = parser.parse_args()
+    parser.add_argument(
+        '--capture', type=str, default=None,
+        help='Capture test data into a file')
+    return parser
 
+
+def main():
+    args = create_parser().parse_args()
     app = CCApplication()
     app['port'] = args.port
     app['oc_port'] = args.oc_port
+    app['protocol_factory'] = protocol
     app.setup_routes()
 
     async def tcp_server(app):
@@ -186,9 +192,38 @@ def main():
         async with server:
             yield
 
-    app.cleanup_ctx.append(tcp_server)
+    async def capture(app):
+        with open(args.capture, 'wb') as f:
+            def protocol_factory(send, sess_cls=sess.CCSession):
+                def write_frame(t, m):
+                    ln = str(len(m)).encode('ascii')
+                    f.write(t + ln + b':' + m + b'\n')
 
-    web.run_app(app, host=args.host, port=args.port)
+                def send_wrap(m):
+                    write_frame(b'S', m)
+                    return send(m)
+
+                p = protocol(send_wrap, sess_cls=sess_cls)
+
+                def pgen():
+                    next(p)
+                    while True:
+                        m = yield
+                        write_frame(b'R', m)
+                        p.send(m)
+
+                return pgen()
+
+            app['protocol_factory'] = protocol_factory
+            yield
+
+    app.cleanup_ctx.append(tcp_server)
+    if args.capture is not None:
+        sess.python_version = lambda: '<VERSION>'
+        app.cleanup_ctx.append(capture)
+
+    with sess.patch_std_files():
+        web.run_app(app, host=args.host, port=args.port)
 
 
 if __name__ == '__main__':

computercraft/sess.py

@@ -138,9 +138,18 @@ def install_import_hook():
 
 
 install_import_hook()
-sys.stdin = StdFileProxy(sys.__stdin__, False)
-sys.stdout = StdFileProxy(sys.__stdout__, False)
-sys.stderr = StdFileProxy(sys.__stderr__, True)
+
+
+@contextmanager
+def patch_std_files():
+    pin, pout, perr = sys.stdin, sys.stdout, sys.stderr
+    sys.stdin = StdFileProxy(pin, False)
+    sys.stdout = StdFileProxy(pout, False)
+    sys.stderr = StdFileProxy(perr, True)
+    try:
+        yield
+    finally:
+        sys.stdin, sys.stdout, sys.stderr = pin, pout, perr
 
 
 def eval_lua(lua_code, *params, immediate=False):

setup.cfg

@@ -1,3 +1,8 @@
 [flake8]
 max-line-length = 120
 ignore = I,C812,N802,N803,N815,N816,W503
+
+[tool:pytest]
+testpaths = tests
+pythonpath = .
+addopts = --import-mode=importlib

tests/conftest.py

@@ -0,0 +1,27 @@
+import pytest
+
+import computercraft.server
+import computercraft.sess
+from computercraft.ser import deserialize
+
+
+@pytest.fixture
+def protocol_tuple():
+    sbuf = []
+
+    def send(m):
+        sbuf.append(deserialize(m))
+
+    pgen = computercraft.server.protocol(sbuf.append)
+    pgen.send(None)
+    return pgen.send, sbuf
+
+
+@pytest.fixture
+def send(protocol_tuple):
+    return protocol_tuple[0]
+
+
+@pytest.fixture
+def buf(protocol_tuple):
+    return protocol_tuple[1]

tests/proto/oc1.8.0_m1.12.2_repl2p2.txt

@@ -0,0 +1,21 @@
+R6:0[4]{}
+S56:T<1>1<20>io.stderr:write(...){:[1]<17>Python <VERSION>
+}
+R18:T<1>1<7>{:[1]T}[1]
+S35:T<1>1<13>io.write(...){:[1]<4>>>> }
+R18:T<1>1<7>{:[1]T}[1]
+S27:T<1>1<16>return io.read(){}
+R29:T<1>1<17>{:[1]T:[2]<3>2+2}[1]
+S32:T<1>1<13>io.write(...){:[1]<1>4}
+R18:T<1>1<7>{:[1]T}[1]
+S32:T<1>1<13>io.write(...){:[1]<1>
+}
+R18:T<1>1<7>{:[1]T}[1]
+S35:T<1>1<13>io.write(...){:[1]<4>>>> }
+R18:T<1>1<7>{:[1]T}[1]
+S27:T<1>1<16>return io.read(){}
+R18:T<1>1<7>{:[1]T}[1]
+S39:T<1>1<20>io.stderr:write(...){:[1]<1>
+}
+R18:T<1>1<7>{:[1]T}[1]
+S2:CN

tests/proto/oc1.8.0_m1.12.2_replterm.txt

@@ -0,0 +1,23 @@
+R6:0[4]{}
+S56:T<1>1<20>io.stderr:write(...){:[1]<17>Python <VERSION>
+}
+R18:T<1>1<7>{:[1]T}[1]
+S35:T<1>1<13>io.write(...){:[1]<4>>>> }
+R18:T<1>1<7>{:[1]T}[1]
+S27:T<1>1<16>return io.read(){}
+R33:T<1>1<21>{:[1]T:[2]<7>5 * 'a'}[1]
+S38:T<1>1<13>io.write(...){:[1]<7>'aaaaa'}
+R18:T<1>1<7>{:[1]T}[1]
+S32:T<1>1<13>io.write(...){:[1]<1>
+}
+R18:T<1>1<7>{:[1]T}[1]
+S35:T<1>1<13>io.write(...){:[1]<4>>>> }
+R18:T<1>1<7>{:[1]T}[1]
+S27:T<1>1<16>return io.read(){}
+R43:T<1>1<31>{:[1]T:[2]F:[3]<11>interrupted}[1]
+S50:I<1>1<20>io.stderr:write(...){:[1]<11>interrupted}
+R13:T<1>1<2>{}[0]
+S39:T<1>1<20>io.stderr:write(...){:[1]<1>
+}
+R18:T<1>1<7>{:[1]T}[1]
+S2:CN

tests/serialization.py

@@ -1,49 +0,0 @@
-from math import inf, nan, isnan
-
-from computercraft.ser import serialize, deserialize
-
-
-roundtrip_vals = [
-    None,
-    True,
-    False,
-    0,
-    -1,
-    1,
-    1e6,
-    1.5,
-    2.4e-9,
-    # nan,
-    inf,
-    -inf,
-    '',
-    'string',
-    '\n\r\0',
-    '\0',
-    '2',
-    {},
-    {2: 4},
-    {'a': 1, 'b': None, 'c': {}, 'd': {'x': 8}},
-    [1, 2, 3],
-    [1],
-    ['abc'],
-]
-
-
-for v in roundtrip_vals:
-    print(serialize(v))
-    assert v == deserialize(serialize(v))
-
-
-print(serialize(nan))
-assert isnan(deserialize(serialize(nan)))
-
-
-oneway_vals = [
-    ({1: 'a', 2: 'b', 3: 'c'}, ['a', 'b', 'c']),
-]
-
-
-for a, b in oneway_vals:
-    print(serialize(a))
-    assert b == deserialize(serialize(a))

tests/test_proto.py

@@ -0,0 +1,53 @@
+from collections import deque
+from pathlib import Path
+
+import pytest
+
+import computercraft.server
+import computercraft.sess
+
+
+@pytest.fixture(autouse=True)
+def _patch(monkeypatch):
+    monkeypatch.setattr(
+        computercraft.sess, 'python_version',
+        lambda: '<VERSION>')
+
+
+_proto_folder = (Path(__file__).parent / 'proto')
+
+
+@pytest.mark.parametrize(
+    'logfile',
+    _proto_folder.glob('**/*.txt'),
+    ids=lambda p: str(p.relative_to(_proto_folder)))
+def test_proto(logfile):
+    sbuf = deque()
+    with computercraft.sess.patch_std_files():
+        pgen = computercraft.server.protocol(sbuf.append)
+        pgen.send(None)
+
+        with logfile.open('rb') as lf:
+
+            def read_frame():
+                flen = []
+                while (b := lf.read(1)) != b':':
+                    flen.append(b)
+                flen = int(b''.join(flen))
+                frame = lf.read(flen)
+                assert len(frame) == flen
+                assert lf.read(1) == b'\n'
+                return frame
+
+            while True:
+                match lf.read(1):
+                    case b'':
+                        break
+                    case b'R':
+                        pgen.send(read_frame())
+                    case b'S':
+                        assert read_frame() == sbuf.popleft()
+                    case _ as x:
+                        raise ValueError('Bad prefix ' + repr(x))
+
+    assert len(sbuf) == 0

tests/test_serialization.py

@@ -0,0 +1,44 @@
+from math import inf, nan, isnan
+
+import pytest
+
+from computercraft.ser import serialize, deserialize
+
+
+@pytest.mark.parametrize('v', [
+    None,
+    True,
+    False,
+    0,
+    -1,
+    1,
+    1e6,
+    1.5,
+    2.4e-9,
+    inf,
+    -inf,
+    b'',
+    b'string',
+    b'\n\r\0',
+    b'\0',
+    b'2',
+    {},
+    {2: 4},
+    {b'a': 1, b'b': None, b'c': {}, b'd': {b'x': 8}},
+])
+def test_roundtrip(v):
+    assert v == deserialize(serialize(v))
+
+
+def test_nan():
+    assert isnan(deserialize(serialize(nan)))
+
+
+@pytest.mark.parametrize('a,b', [
+    ([1], {1: 1}),
+    ([1, 2, 3], {1: 1, 2: 2, 3: 3}),
+    ([b'abc'], {1: b'abc'}),
+    ([b'a', b'b', b'c'], {1: b'a', 2: b'b', 3: b'c'}),
+])
+def test_oneway(a, b):
+    assert b == deserialize(serialize(a))