Initial work with tests for message handling in Python.

ntoll · ntoll · commit e4593cb5a372 · 2022-12-15T18:11:53.000Z
diff --git a/SpecRunner.html b/SpecRunner.html
@@ -21,7 +21,7 @@
   <body>
     <div style="display: none;">
       <p id="testParaID">This is a test paragraph, with an id.</p>
-      <p class="testParaClass">This is a test paragrapch, with a class.</p>
+      <p class="testParaClass">This is a test paragraph, with a class.</p>
       <button id="testButton">Click Me</button>
       <button id="testButton2">Click Me Again</button>
       <div id="testMutate"></div>
diff --git a/polyplug.py b/polyplug.py
@@ -1,8 +1,11 @@
+import binascii
+import builtins
+import hashlib
 import json
 
 
-def output(content):
-    print(json.dumps({"type": "stdout", "content": content}))
+# Registered event listeners.
+LISTENERS = {}
 
 
 class Query:
@@ -525,7 +528,7 @@ def _find_by_tagName(self, target):
         ):
             result.extend(child._find_by_tagName(target))
         return result
-        
+
 
 class TextNode(Node):
     """
@@ -608,6 +611,62 @@ def as_dict(self):
         return {"nodeType": 11, "childNodes": []}
 
 
+def get_listener_id(query, event_type, listener):
+    """
+    Given a query, event type and listener function, generate a unique id from
+    this combination.
+    """
+    raw = str(query.as_dict) + event_type + listener.__name__
+    return binascii.hexlify(
+        hashlib.sha256(raw.encode("utf-8")).digest()
+    ).decode("ascii")
+
+
+def print(*args, **kwargs):
+    """
+    Overridden print so output is handled correctly via JSON message passing
+    instead of just raw text.
+    """
+    sep = kwargs.get("sep", " ")
+    end = kwargs.get("end", "\n")
+    content = sep.join(args) + end
+    builtins.print(json.dumps({"type": "stdout", "content": content}))
+
+
+def update(query, target):
+    """
+    Update the DOM so the node[s] matching the query are mutated to the state
+    defined by the target node.
+    """
+    builtins.print(
+        json.dumps(
+            {
+                "type": "updateDOM",
+                "query": query.as_dict,
+                "target": target.as_dict,
+            }
+        )
+    )
+
+
+def remove(query, event_type, listener):
+    """
+    Remove the referenced listener from handling the event_type from the
+    node[s] matching the query.
+    """
+    listener_id = get_listener_id(query, event_type, listener)
+    del LISTENERS[listener_id]
+    builtins.print(
+        json.dumps(
+            {
+                "type": "removeEvent",
+                "query": query.as_dict,
+                "eventType": event_type,
+            }
+        )
+    )
+
+
 def plug(query, event_type):
     """
     A decorator wrapper to plug a Python function into a DOM event specified
@@ -622,10 +681,56 @@ def plug(query, event_type):
     """
 
     def decorator(fn):
-        @wraps(fn)
-        def wrapper(*args, **kwargs):
-            fn(*args, **kwargs)
+        """
+        Register the function via the query and event_type.
+        """
+
+        def wrapper(event):
+            return fn(event)
+
+        listener_id = get_listener_id(query, event_type, wrapper)
+        builtins.print(
+            json.dumps(
+                {
+                    "type": "registerEvent",
+                    "query": query.as_dict,
+                    "eventType": event_type,
+                    "listener": listener_id,
+                }
+            )
+        )
 
+        LISTENERS[listener_id] = wrapper
         return wrapper
 
     return decorator
+
+
+def receive(raw):
+    """
+    Given a raw JSON message, decode it, find the expected handler function,
+    re-constitute the DOM, and call the handler with the appropriate context.
+    """
+    try:
+        msg = json.loads(raw)
+        event_type = msg.get("type")
+        target = msg.get("target")
+        listener = msg.get("listener")
+        if event_type and target and listener:
+            if listener in LISTENERS:
+                event = DomEvent(event_type, ElementNode(**target))
+                LISTENERS[listener](event)
+            else:
+                raise RuntimeError("No such listener: " + listener)
+        else:
+            raise ValueError("Incomplete message received: " + raw)
+    except Exception as ex:
+        context = {"type": type(ex).__name__, "msg": str(ex)}
+        builtins.print(
+            json.dumps(
+                {
+                    "type": "error",
+                    "context": context,
+                }
+            )
+        )
diff --git a/tests/test_polyplug.py b/tests/test_polyplug.py
@@ -1,13 +1,27 @@
 """
 Exercise PolyPlug.
 """
+import builtins
 import copy
 import json
 import pytest
 import polyplug
 from unittest import mock
 
 
+@pytest.fixture(autouse=True)
+def test_wrapper():
+    """
+    Ensures clean state.
+    """
+    # Clear the listeners.
+    polyplug.LISTENERS = {}
+    # Run the test.
+    yield
+    # ???
+    # Profit!
+
+
 DOM_FROM_JSON = {
     "nodeType": 1,
     "tagName": "div",
@@ -892,3 +906,203 @@ def test_htmltokenizer_tokenize_complex_tree():
     tok = polyplug.HTMLTokenizer(raw)
     tok.tokenize(parent)
     assert parent.as_dict == expected
+
+
+def test_get_listener_id():
+    """
+    Return a string containing a hex representation of a sha256 hash of the
+    passed in Query, event type and listener function.
+    """
+    q = polyplug.Query(id="foo")
+    event_type = "click"
+
+    def test_fn():
+        pass
+
+    id_1 = polyplug.get_listener_id(q, event_type, test_fn)
+    id_2 = polyplug.get_listener_id(q, event_type, test_fn)
+    assert id_1 == id_2  # These should be the same..!
+
+
+def test_print():
+    """
+    The polyplug print function emits the expected JSON message.
+    """
+    with mock.patch("builtins.print") as mock_print:
+        # Simple case with defaults.
+        polyplug.print("Hello", "world")
+        mock_print.assert_called_once_with(
+            '{"type": "stdout", "content": "Hello world\\n"}'
+        )
+        mock_print.reset_mock()
+        # More complex with sep and end
+        polyplug.print("Hello", "world", sep="-", end="")
+        mock_print.assert_called_once_with(
+            '{"type": "stdout", "content": "Hello-world"}'
+        )
+
+
+def test_update():
+    """
+    Given a query object and a representation of a target node, the expected
+    updateDOM message is emitted, with the correct payload.
+    """
+    query = polyplug.Query(id="foo")
+    raw_dom = copy.deepcopy(DOM_FROM_JSON)
+    target = polyplug.ElementNode(**raw_dom)
+    with mock.patch("builtins.print") as mock_print:
+        polyplug.update(query, target)
+        assert mock_print.call_count == 1
+        msg = json.loads(mock_print.call_args.args[0])
+        assert msg["type"] == "updateDOM"
+        assert msg["query"]["id"] == "foo"
+        assert msg["target"] == DOM_FROM_JSON
+
+
+def test_remove():
+    """ """
+    with mock.patch("builtins.print") as mock_print:
+
+        @polyplug.plug(polyplug.Query(id="foo"), "some-event")
+        def test_fn(event):
+            return "It works!"
+
+        assert mock_print.call_count == 1
+        listener_id = polyplug.get_listener_id(
+            polyplug.Query(id="foo"), "some-event", test_fn
+        )
+        assert listener_id in polyplug.LISTENERS
+        mock_print.reset_mock()
+        polyplug.remove(polyplug.Query(id="foo"), "some-event", test_fn)
+        assert mock_print.call_count == 1
+        msg = json.loads(mock_print.call_args.args[0])
+        assert msg["type"] == "removeEvent"
+        assert msg["query"]["id"] == "foo"
+        assert msg["eventType"] == "some-event"
+        assert listener_id not in polyplug.LISTENERS
+
+
+def test_plug_decorator_register():
+    """
+    Ensure the expected register JSON message is emitted when the decorator is
+    used on a user's function.
+    """
+    with mock.patch("builtins.print") as mock_print:
+
+        @polyplug.plug(polyplug.Query(id="foo"), "some-event")
+        def test_fn(event):
+            return "It works!"
+
+        assert mock_print.call_count == 1
+        msg = json.loads(mock_print.call_args.args[0])
+        assert msg["type"] == "registerEvent"
+        assert msg["listener"] == polyplug.get_listener_id(
+            polyplug.Query(id="foo"), "some-event", test_fn
+        )
+        assert msg["query"]["id"] == "foo"
+        assert msg["eventType"] == "some-event"
+        result = polyplug.LISTENERS[msg["listener"]](None)
+        assert result == "It works!"
+
+
+def test_receive_bad_json():
+    """
+    If the receive function get a non-JSON message, it complains with an error
+    message of its own.
+    """
+    with mock.patch("builtins.print") as mock_print:
+        polyplug.receive("not VALID")
+        assert mock_print.call_count == 1
+        msg = json.loads(mock_print.call_args.args[0])
+        assert msg["type"] == "error"
+        assert msg["context"]["type"] == "JSONDecodeError"
+        assert (
+            msg["context"]["msg"]
+            == "Expecting value: line 1 column 1 (char 0)"
+        )
+
+
+def test_receive_incomplete_message():
+    """
+    If the receive function gets valid JSON that is the wrong "shape", it
+    complains with a message of its own.
+    """
+    with mock.patch("builtins.print") as mock_print:
+        polyplug.receive(json.dumps({"foo": "bar"}))
+        assert mock_print.call_count == 1
+        msg = json.loads(mock_print.call_args.args[0])
+        assert msg["type"] == "error"
+        assert msg["context"]["type"] == "ValueError"
+        assert (
+            msg["context"]["msg"]
+            == 'Incomplete message received: {"foo": "bar"}'
+        )
+
+
+def test_receive_no_listener():
+    """
+    If the receive function gets a valid message but the referenced listener
+    function doesn't exist, it complains with a message of its own.
+    """
+    with mock.patch("builtins.print") as mock_print:
+        polyplug.receive(
+            json.dumps(
+                {
+                    "type": "some-event",
+                    "target": DOM_FROM_JSON,
+                    "listener": "does_not_exist",
+                }
+            )
+        )
+        assert mock_print.call_count == 1
+        msg = json.loads(mock_print.call_args.args[0])
+        assert msg["type"] == "error"
+        assert msg["context"]["type"] == "RuntimeError"
+        assert msg["context"]["msg"] == "No such listener: does_not_exist"
+
+
+def test_receive_for_registered_listener():
+    """
+    If the receive function gets a valid message for an existing event,
+    the function is called with the expected DomEvent object.
+    """
+    with mock.patch("builtins.print") as mock_print:
+        # To be called when there's user defined work to be done.
+        mock_work = mock.MagicMock()
+
+        @polyplug.plug(polyplug.Query(id="foo"), "some-event")
+        def test_fn(event):
+            """
+            Do some work as if an end user.
+            """
+            # Expected eventType.
+            if event.event_type != "some-event":
+                raise ValueError("It broke! Wrong event.")
+            # The target represents the expected element.
+            if event.target.tagName != "div":
+                raise ValueError("It broke! Wrong target root.")
+            # It's possible to find one of the expected child nodes.
+            ul = event.target.find(".list")
+            if ul.tagName != "ul":
+                raise ValueError("It broke! Wrong child nodes.")
+            # Signal things worked out. ;-)
+            mock_work("It works!")
+
+        assert mock_print.call_count == 1
+        mock_print.reset_mock()
+
+        listener_id = polyplug.get_listener_id(
+            polyplug.Query(id="foo"), "some-event", test_fn
+        )
+
+        polyplug.receive(
+            json.dumps(
+                {
+                    "type": "some-event",
+                    "target": DOM_FROM_JSON,
+                    "listener": listener_id,
+                }
+            )
+        )
+
+        mock_work.assert_called_once_with("It works!")
