add test file and modified unittest

brentru · brentru · commit 1f87bdd29e1e · 2019-07-17T13:03:39.000-04:00
diff --git a/examples/tests/adafruit_io_http_tester.py b/examples/tests/adafruit_io_http_tester.py
@@ -0,0 +1,142 @@
+"""
+CircuitPython_AdafruitIO IO_HTTP Tester
+--------------------------------------------------
+
+Tests Adafruit IO CircuitPython HTTP method
+coverage with a WiFi CircuitPython device.
+
+* Author(s): Brent Rubell for Adafruit Industries
+"""
+from random import randint, uniform
+import time
+import board
+import busio
+from digitalio import DigitalInOut
+from adafruit_esp32spi import adafruit_esp32spi, adafruit_esp32spi_wifimanager
+import neopixel
+from adafruit_io.adafruit_io import IO_HTTP, AdafruitIO_RequestError
+
+# REQUIRES MicroPython's UnitTest
+# https://github.com/micropython/micropython-lib/tree/master/unittest
+import unittest
+
+# Get wifi details and more from a secrets.py file
+try:
+    from secrets import secrets
+except ImportError:
+    print("WiFi secrets are kept in secrets.py, please add them there!")
+    raise
+
+# ESP32 Setup
+esp32_cs = DigitalInOut(board.ESP_CS)
+esp32_ready = DigitalInOut(board.ESP_BUSY)
+esp32_reset = DigitalInOut(board.ESP_RESET)
+spi = busio.SPI(board.SCK, board.MOSI, board.MISO)
+esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
+status_light = neopixel.NeoPixel(board.NEOPIXEL, 1, brightness=0.2)
+wifi = adafruit_esp32spi_wifimanager.ESPSPI_WiFiManager(esp, secrets, status_light)
+
+# Set your Adafruit IO Username and Key in secrets.py
+# (visit io.adafruit.com if you need to create an account,
+# or if you need your Adafruit IO key.)
+aio_username = secrets["aio_user"]
+aio_key = secrets["aio_password"]
+
+
+class Test_IO_HTTP(unittest.TestCase):
+
+    # Tests for Adafruit IO Authentication
+    def test_set_user_key(self):
+        """__init__ constructor
+        correctly exposes provided credentials.
+        """
+        username = "adabot"
+        key = "mho"
+        io = IO_HTTP(username, key, wifi)
+        self.assertEqual(username, io.username)
+        self.assertEqual(key, io.key)
+
+    def test_incorrect_user_pass_action(self):
+        """Incorrect credentials provided to __init__
+        should raise a RequestError.
+        """
+        username = "adabot"
+        key = "mho"
+        io = IO_HTTP(username, key, wifi)
+        with self.assertRaises(AdafruitIO_RequestError):
+            test_feed = io.get_feed("errorfeed")
+            pass
+
+    # Tests for Adafruit IO Data Methods
+    def test_txrx(self):
+        """Sends a random integer value to a feed and receives it back.
+        """
+        # Create an Adafruit IO HTTP Client
+        io = IO_HTTP(aio_username, aio_key, wifi)
+        try:
+            test_feed = io.get_feed("testfeed")
+        except AdafruitIO_RequestError:
+            test_feed = io.create_new_feed("testfeed")
+        tx_data = randint(1, 100)
+        # send the value
+        io.send_data(test_feed["key"], tx_data)
+        # and get it back...
+        rx_data = io.receive_data(test_feed["key"])
+        self.assertEqual(int(rx_data["value"]), tx_data)
+
+    def test_send_location_data(self):
+        """Sets location metadata.
+        send_data
+        """
+        # Create an Adafruit IO HTTP Client
+        io = IO_HTTP(aio_username, aio_key, wifi)
+        io.delete_feed('testfeed')
+        test_feed = io.create_new_feed('testfeed')
+        # value
+        value = randint(1, 100)
+        # Set up metadata associated with value
+        metadata = {'lat': uniform(1, 100),
+                    'lon': uniform(1, 100),
+                    'ele': 10,
+                    'created_at': None}
+        io.send_data(test_feed['key'], value, metadata)
+        rx_data = io.receive_data(test_feed['key'])
+        self.assertEqual(int(rx_data['value']), value)
+        self.assertAlmostEqual(float(rx_data['lat']), metadata['lat'])
+        self.assertAlmostEqual(float(rx_data['lon']), metadata['lon'])
+        self.assertAlmostEqual(float(rx_data['ele']), metadata['ele'])
+
+    # Test for Adafruit IO Feed Methods
+    def test_create_feed(self):
+        """Test creating a new feed.
+        """
+        # Create an Adafruit IO HTTP Client
+        io = IO_HTTP(aio_username, aio_key, wifi)
+        io.delete_feed('testfeed')
+        test_feed = io.create_new_feed('testfeed')
+        self.assertEqual(test_feed['name'], 'testfeed')
+
+    def test_delete_feed(self):
+        """delete_feed by feed key
+        """
+        # Create an Adafruit IO HTTP Client
+        io = IO_HTTP(aio_username, aio_key, wifi)
+        io.delete_feed('testfeed')
+        with self.assertRaises(AdafruitIO_RequestError):
+            io.receive_data('testfeed'['key'])
+            pass
+
+    def test_delete_nonexistent_feed(self):
+        """delete nonexistent feed by feed key
+        """
+        # Create an Adafruit IO HTTP Client
+        io = IO_HTTP(aio_username, aio_key, wifi)
+        io.delete_feed('testfeed')
+        with self.assertRaises(AdafruitIO_RequestError):
+            io.delete_feed['testfeed']
+
+
+if __name__ == "__main__":
+    # Pass the NetworkManager Object to UnitTest.py
+    unittest.get_wifi(wifi)
+    unittest.main()
diff --git a/examples/tests/unittest.py b/examples/tests/unittest.py
@@ -0,0 +1,233 @@
+# UnitTest.py
+# https://github.com/micropython/micropython-lib/blob/master/unittest/unittest.py
+# Modified for handling ESP32SPI module connectivity
+
+class SkipTest(Exception):
+    pass
+
+
+class AssertRaisesContext:
+
+    def __init__(self, exc):
+        self.expected = exc
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, tb):
+        if exc_type is None:
+            assert False, "%r not raised" % self.expected
+        if issubclass(exc_type, self.expected):
+            return True
+        return False
+
+
+class TestCase:
+
+    def fail(self, msg=''):
+        assert False, msg
+
+    def assertEqual(self, x, y, msg=''):
+        if not msg:
+            msg = "%r vs (expected) %r" % (x, y)
+        assert x == y, msg
+
+    def assertNotEqual(self, x, y, msg=''):
+        if not msg:
+            msg = "%r not expected to be equal %r" % (x, y)
+        assert x != y, msg
+
+    def assertAlmostEqual(self, x, y, places=None, msg='', delta=None):
+        if x == y:
+            return
+        if delta is not None and places is not None:
+            raise TypeError("specify delta or places not both")
+
+        if delta is not None:
+            if abs(x - y) <= delta:
+                return
+            if not msg:
+                msg = '%r != %r within %r delta' % (x, y, delta)
+        else:
+            if places is None:
+                places = 7
+            if round(abs(y-x), places) == 0:
+                return
+            if not msg:
+                msg = '%r != %r within %r places' % (x, y, places)
+
+        assert False, msg
+
+    def assertNotAlmostEqual(self, x, y, places=None, msg='', delta=None):
+        if delta is not None and places is not None:
+            raise TypeError("specify delta or places not both")
+
+        if delta is not None:
+            if not (x == y) and abs(x - y) > delta:
+                return
+            if not msg:
+                msg = '%r == %r within %r delta' % (x, y, delta)
+        else:
+            if places is None:
+                places = 7
+            if not (x == y) and round(abs(y-x), places) != 0:
+                return
+            if not msg:
+                msg = '%r == %r within %r places' % (x, y, places)
+
+        assert False, msg
+
+    def assertIs(self, x, y, msg=''):
+        if not msg:
+            msg = "%r is not %r" % (x, y)
+        assert x is y, msg
+
+    def assertIsNot(self, x, y, msg=''):
+        if not msg:
+            msg = "%r is %r" % (x, y)
+        assert x is not y, msg
+
+    def assertIsNone(self, x, msg=''):
+        if not msg:
+            msg = "%r is not None" % x
+        assert x is None, msg
+
+    def assertIsNotNone(self, x, msg=''):
+        if not msg:
+            msg = "%r is None" % x
+        assert x is not None, msg
+
+    def assertTrue(self, x, msg=''):
+        if not msg:
+            msg = "Expected %r to be True" % x
+        assert x, msg
+
+    def assertFalse(self, x, msg=''):
+        if not msg:
+            msg = "Expected %r to be False" % x
+        assert not x, msg
+
+    def assertIn(self, x, y, msg=''):
+        if not msg:
+            msg = "Expected %r to be in %r" % (x, y)
+        assert x in y, msg
+
+    def assertIsInstance(self, x, y, msg=''):
+        assert isinstance(x, y), msg
+
+    def assertRaises(self, exc, func=None, *args, **kwargs):
+        if func is None:
+            return AssertRaisesContext(exc)
+
+        try:
+            func(*args, **kwargs)
+            assert False, "%r not raised" % exc
+        except Exception as e:
+            if isinstance(e, exc):
+                return
+            raise
+
+
+
+def skip(msg):
+    def _decor(fun):
+        # We just replace original fun with _inner
+        def _inner(self):
+            raise SkipTest(msg)
+        return _inner
+    return _decor
+
+def skipIf(cond, msg):
+    if not cond:
+        return lambda x: x
+    return skip(msg)
+
+def skipUnless(cond, msg):
+    if cond:
+        return lambda x: x
+    return skip(msg)
+
+
+class TestSuite:
+    def __init__(self):
+        self.tests = []
+    def addTest(self, cls):
+        self.tests.append(cls)
+
+class TestRunner:
+    def run(self, suite):
+        res = TestResult()
+        for c in suite.tests:
+            run_class(c, res)
+
+        print("Ran %d tests\n" % res.testsRun)
+        if res.failuresNum > 0 or res.errorsNum > 0:
+            print("FAILED (failures=%d, errors=%d)" % (res.failuresNum, res.errorsNum))
+        else:
+            msg = "OK"
+            if res.skippedNum > 0:
+                msg += " (%d skipped)" % res.skippedNum
+            print(msg)
+
+        return res
+
+class TestResult:
+    def __init__(self):
+        self.errorsNum = 0
+        self.failuresNum = 0
+        self.skippedNum = 0
+        self.testsRun = 0
+
+    def wasSuccessful(self):
+        return self.errorsNum == 0 and self.failuresNum == 0
+
+# TODO: Uncompliant
+def run_class(c, test_result):
+    o = c()
+    set_up = getattr(o, "setUp", lambda: None)
+    tear_down = getattr(o, "tearDown", lambda: None)
+    for name in dir(o):
+        if name.startswith("test"):
+            print("%s (%s) ..." % (name, c.__qualname__), end="")
+            m = getattr(o, name)
+            set_up()
+            try:
+                test_result.testsRun += 1
+                m()
+                print(" ok")
+            except SkipTest as e:
+                print(" skipped:", e.args[0])
+                test_result.skippedNum += 1
+            except (RuntimeError, ValueError) as e:
+                print('Failed to get data from ESP32, retrying...\n')
+                wifi.reset()
+                pass
+            except:
+                print(" FAIL")
+                test_result.failuresNum += 1
+                # Uncomment to investigate failure in detail
+                raise
+                continue
+            finally:
+                tear_down()
+
+def get_wifi(wifi_module):
+    wifi = wifi_module
+    return wifi
+
+def main(module="__main__"):
+    def test_cases(m):
+        for tn in dir(m):
+            c = getattr(m, tn)
+            if isinstance(c, object) and isinstance(c, type) and issubclass(c, TestCase):
+                yield c
+
+    m = __import__(module)
+    suite = TestSuite()
+    for c in test_cases(m):
+        suite.addTest(c)
+    runner = TestRunner()
+    result = runner.run(suite)
+    # Terminate with non zero return code in case of failures
+    if result.failuresNum > 0:
+        raise ValueError('Failures: ', result.failuresNum)
