Add support and tests for read/write

tests/conftest.py

@@ -12,6 +12,15 @@ def jsonFunc():
         self.json = jsonFunc
 
 
+class MockedRead:
+    def __init__(self, response):
+        def jsonFunc():
+            return response
+
+        self.ok = True
+        self.json = jsonFunc
+
+
 def create_fake_cred_file():
     with open(
         os.path.dirname(os.path.dirname(__file__)) + "/credentials.txt", "w"
@@ -62,8 +71,35 @@ def no_token_file():
 
 
 @pytest.fixture
-def patched_requests(monkeypatch):
+def patched_post(monkeypatch):
     def mocked_post(uri, *args, **kwargs):
         return MockedResponse()
 
     monkeypatch.setattr(requests, "post", mocked_post)
+
+
+@pytest.fixture
+def patched_get(monkeypatch):
+    def mocked_get(uri, *args, **kwargs):
+        if uri == "https://rw.vestaboard.com":
+            return MockedRead(
+                {
+                    "currentMessage": {
+                        "layout": "[[1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[2,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[3,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[4,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[5,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[6,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]]"
+                    }
+                }
+            )
+        return MockedRead(
+            {
+                "message": [
+                    [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+                    [2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+                    [3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+                    [4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+                    [5, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+                    [6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+                ]
+            }
+        )
+
+    monkeypatch.setattr(requests, "get", mocked_get)

tests/test_localApi.py

@@ -2,7 +2,7 @@
 import vestaboard
 
 
-def test_enabling_local_api_succeeds(patched_requests):
+def test_enabling_local_api_succeeds(patched_post):
     b = vestaboard.Board(localApi={"enablementToken": "fake", "ip": "10.0.0.1"})
     assert b.localOptions["useSavedToken"] == True
 
@@ -19,7 +19,7 @@ def test_using_saved_key_and_ip(with_token_file):
     assert b.localIP == "10.0.0.2"
 
 
-def test_warns_when_both_key_and_enablement_token(patched_requests):
+def test_warns_when_both_key_and_enablement_token(patched_post):
     with pytest.warns(
         UserWarning,
         match="An enablement token was provided, so I'm enabling the local API for you. If you've already enabled the local API on your board, remove the enablement token and try again.",
@@ -61,3 +61,47 @@ def test_nonexistent_token_file(no_token_file):
 def test_fails_with_ip_only(no_token_file):
     with pytest.raises(ValueError):
         vestaboard.Board(localApi={"ip": "10.0.0.1"})
+
+
+def test_should_allow_reading(with_token_file, patched_get):
+    b = vestaboard.Board(localApi={"useSavedToken": True})
+    chars = b.read()
+    assert chars == {
+        "message": [
+            [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            [2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            [3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            [4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            [5, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            [6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+        ]
+    }
+
+
+def test_converts_back_into_characters(with_token_file, patched_get):
+    b = vestaboard.Board(localApi={"useSavedToken": True})
+    chars = b.read({"convert": True})
+    # fmt: off
+    assert chars == [
+                    ['a', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['b', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['c', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['d', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['e', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['f', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                ]
+    # fmt: on
+
+
+def test_joins_back_into_lines_with_normalize(with_token_file, patched_get):
+    b = vestaboard.Board(localApi={"useSavedToken": True})
+    chars = b.read({"convert": True, "normalize": True})
+    assert (
+        chars
+        == "a                     \n"
+        + "b                     \n"
+        + "c                     \n"
+        + "d                     \n"
+        + "e                     \n"
+        + "f                     "
+    )

tests/test_readWrite.py

@@ -0,0 +1,53 @@
+import re
+import pytest
+import vestaboard
+
+
+def test_instantiating_read_write_board():
+    b = vestaboard.Board(apiKey="my read write key", readWrite=True)
+    assert b.apiKey == "my read write key", "Should use the passed-in API key"
+
+
+def test_should_allow_reading(no_credentials, patched_get):
+    b = vestaboard.Board(apiKey="my read write key", readWrite=True)
+    chars = b.read()
+    assert chars == {
+        "currentMessage": {
+            "layout": "[[1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[2,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[3,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[4,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[5,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[6,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]]"
+        }
+    }
+
+
+def test_converts_back_into_characters(no_credentials, patched_get):
+    b = vestaboard.Board(apiKey="my read write key", readWrite=True)
+    chars = b.read({"convert": True})
+    # fmt: off
+    assert chars == [
+                    ['a', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['b', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['c', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['d', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['e', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                    ['f', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' '],
+                ]
+    # fmt: on
+
+
+def test_joins_back_into_lines_with_normalize(no_credentials, patched_get):
+    b = vestaboard.Board(apiKey="my read write key", readWrite=True)
+    chars = b.read({"convert": True, "normalize": True})
+    assert (
+        chars
+        == "a                     \n"
+        + "b                     \n"
+        + "c                     \n"
+        + "d                     \n"
+        + "e                     \n"
+        + "f                     "
+    )
+
+
+def test_disallows_reading_on_regular_board(with_credentials):
+    with pytest.raises(ValueError, match=re.escape('.read() is only available when using local API or by using a read/write enabled API key.\nPass "readWrite=True" along with your apiKey to enable readWrite mode.')):
+        b = vestaboard.Board()
+        b.read()

vestaboard/__init__.py

@@ -6,8 +6,6 @@
 Board - Class
 Installable - Class
 """
-from cgitb import enable
-from multiprocessing.sharedctypes import Value
 import requests
 from vestaboard.formatter import Formatter
 import vestaboard.vbUrls as vbUrls
@@ -24,6 +22,7 @@ def __init__(
         apiSecret=False,
         subscriptionId=False,
         localApi: dict = None,
+        readWrite: bool = False,
     ):
         """
         Returns an instance of Board().
@@ -53,6 +52,7 @@ def __init__(
         self.localOptions = localApi
         self.localKey = ""
         self.localIP = ""
+        self.readWrite = readWrite
 
         if self.localOptions is not None:
             if self.checkAndEnableLocalAPI():
@@ -75,6 +75,23 @@ def __init__(
             return
 
         if not installable:  # check for cred file
+            if readWrite:
+                if not apiKey:
+                    try:
+                        creds = get_creds()
+                        if len(creds) < 2:
+                            raise ValueError(
+                                "You must supply a read/write enabled API key to use readWrite mode, or instantiate an Installable in readWrite mode to store your credentials."
+                            )
+                        else:
+                            self.apiKey = creds[0]
+                    except FileNotFoundError as e:
+                        raise FileNotFoundError(
+                            "I couldn't find any saved credentials, and no API key was provided.\nTo use readWrite mode, you must supply a read/write enabled API key, or instantiate an Installable in readWrite mode."
+                        ) from e
+                else:
+                    self.apiKey = apiKey
+                    return
             if not apiKey or not apiSecret or not subscriptionId:
                 try:
                     creds = get_creds()
@@ -86,10 +103,10 @@ def __init__(
                         raise ValueError(
                             "Credentials have been saved, but one or more are missing. Create a new installable and pass in saveCredentials=True, or pass in all three parameters when initializing a new Board."
                         )
-                except FileNotFoundError:
+                except FileNotFoundError as e:
                     raise ValueError(
                         "You must create an installable first or save credentials by passing saveCredentials=True into installable()."
-                    )
+                    ) from e
             else:
                 self.apiKey = apiKey
                 self.apiSecret = apiSecret
@@ -161,6 +178,13 @@ def raw(self, charList: list, pad=None):
         finalText = Formatter()._raw(charList)
         if self.localKey and self.localIP:
             self._raw_local(finalText["characters"])
+        elif self.readWrite and self.apiKey:
+            headers = {"X-Vestaboard-Read-Write-Key": self.apiKey}
+            requests.post(
+                vbUrls.readWrite,
+                headers=headers,
+                data=json.dumps(finalText["characters"]),
+            )
         else:
             headers = {
                 "X-Vestaboard-Api-Key": self.apiKey,
@@ -228,24 +252,31 @@ def checkAndEnableLocalAPI(self):
         return True
 
     def read(self, options: dict = {}):
+        if not self.readWrite:
+            if self.localOptions is None or "useSavedToken" not in self.localOptions:
+                raise ValueError(
+                    '.read() is only available when using local API or by using a read/write enabled API key.\nPass "readWrite=True" along with your apiKey to enable readWrite mode.'
+                )
         if self.localIP and self.localKey:
             localHeader = {"X-Vestaboard-Local-Api-Key": self.localKey}
             res = requests.get(
                 vbUrls.postLocal.format(self.localIP), headers=localHeader
             )
-            if "print" in options and options["print"]:
-                print(res.json())
-            if "convert" in options and options["convert"]:
-                if "normalize" in options and options["normalize"]:
-                    converted = Formatter()._reverse_convert(
-                        res.json()["message"], normalize=True
-                    )
-                    return converted
-                else:
-                    converted = Formatter()._reverse_convert(res.json()["message"])
-                    return converted
-            return res.json()
-        return 'Standard boards do not support reading messages.'
+            response_text = res.json()["message"]
+        elif self.readWrite and self.apiKey:
+            readWriteHeader = {"X-Vestaboard-Read-Write-Key": self.apiKey}
+            res = requests.get(vbUrls.readWrite, headers=readWriteHeader)
+            response_text = json.loads(res.json()["currentMessage"]["layout"])
+        if "print" in options and options["print"]:
+            print(res.json())
+        if "convert" in options and options["convert"]:
+            if "normalize" in options and options["normalize"]:
+                converted = Formatter()._reverse_convert(response_text, normalize=True)
+                return converted
+            else:
+                converted = Formatter()._reverse_convert(response_text)
+                return converted
+        return res.json()
 
     def _post_local(self, text):
         print(