# Test BLE GAP connect/disconnect

from micropython import const
import time, machine, bluetooth

TIMEOUT_MS = 5000

_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)

            

Reported by Pylint.

              # Test BLE GAP connect/disconnect

from micropython import const
import time, machine, bluetooth

TIMEOUT_MS = 5000

_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)

            

Reported by Pylint.

              # Test BLE GAP connect/disconnect

from micropython import const
import time, machine, bluetooth

TIMEOUT_MS = 5000

_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)

            

Reported by Pylint.

              

def wait_for_event(event, timeout_ms):
    t0 = time.ticks_ms()
    while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
        if event in waiting_events:
            return waiting_events.pop(event)
        machine.idle()
    raise ValueError("Timeout waiting for {}".format(event))

            

Reported by Pylint.

              
def wait_for_event(event, timeout_ms):
    t0 = time.ticks_ms()
    while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
        if event in waiting_events:
            return waiting_events.pop(event)
        machine.idle()
    raise ValueError("Timeout waiting for {}".format(event))


            

Reported by Pylint.

              
def wait_for_event(event, timeout_ms):
    t0 = time.ticks_ms()
    while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
        if event in waiting_events:
            return waiting_events.pop(event)
        machine.idle()
    raise ValueError("Timeout waiting for {}".format(event))


            

Reported by Pylint.

              
# Acting in peripheral role.
def instance0():
    multitest.globals(BDADDR=ble.config("mac"))
    ble.gatts_register_services(SERVICES)
    print("gap_advertise")
    ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY")
    multitest.next()
    try:

            

Reported by Pylint.

                  ble.gatts_register_services(SERVICES)
    print("gap_advertise")
    ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY")
    multitest.next()
    try:
        wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
        wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS)
    finally:
        ble.active(0)

            

Reported by Pylint.

              
# Acting in central role.
def instance1():
    multitest.next()
    try:
        # Connect to peripheral and then disconnect.
        print("gap_connect")
        ble.gap_connect(*BDADDR)
        conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)

            

Reported by Pylint.

                  try:
        # Connect to peripheral and then disconnect.
        print("gap_connect")
        ble.gap_connect(*BDADDR)
        conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)

        # Discover services.
        ble.gattc_discover_services(conn_handle)
        wait_for_event(_IRQ_GATTC_SERVICE_DONE, TIMEOUT_MS)

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, uerrno as errno, sys
except:
    import socket, ssl, errno, sys, time, select


def test_one(site, opts):
    ai = socket.getaddrinfo(site, 443)
    addr = ai[0][-1]

            

Reported by Pylint.

                  import socket, ssl, errno, sys, time, select


def test_one(site, opts):
    ai = socket.getaddrinfo(site, 443)
    addr = ai[0][-1]
    print(addr)

    # Connect the raw socket

            

Reported by Pylint.

                      # Wrap with SSL
        try:
            if sys.implementation.name == "micropython":
                s = ssl.wrap_socket(s, do_handshake=False)
            else:
                s = ssl.wrap_socket(s, do_handshake_on_connect=False)
        except OSError as e:
            if e.errno != errno.EINPROGRESS:
                raise

            

Reported by Pylint.

                          if sys.implementation.name == "micropython":
                s = ssl.wrap_socket(s, do_handshake=False)
            else:
                s = ssl.wrap_socket(s, do_handshake_on_connect=False)
        except OSError as e:
            if e.errno != errno.EINPROGRESS:
                raise
        print("wrapped")


            

Reported by Pylint.

                      try:
            test_one(site, opts)
            print(site, "ok")
        except Exception as e:
            print(site, "error")
    print("DONE")


main()

            

Reported by Pylint.

                      try:
            test_one(site, opts)
            print(site, "ok")
        except Exception as e:
            print(site, "error")
    print("DONE")


main()

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, uerrno as errno, sys
except:
    import socket, ssl, errno, sys, time, select


def test_one(site, opts):
    ai = socket.getaddrinfo(site, 443)
    addr = ai[0][-1]

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, uerrno as errno, sys
except:
    import socket, ssl, errno, sys, time, select


def test_one(site, opts):
    ai = socket.getaddrinfo(site, 443)
    addr = ai[0][-1]

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, uerrno as errno, sys
except:
    import socket, ssl, errno, sys, time, select


def test_one(site, opts):
    ai = socket.getaddrinfo(site, 443)
    addr = ai[0][-1]

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, uerrno as errno, sys
except:
    import socket, ssl, errno, sys, time, select


def test_one(site, opts):
    ai = socket.getaddrinfo(site, 443)
    addr = ai[0][-1]

            

Reported by Pylint.

              # test bl and bx instructions


@micropython.asm_thumb
def f(r0):
    # jump over the internal functions
    b(entry)

    label(func1)

            

Reported by Pylint.

              @micropython.asm_thumb
def f(r0):
    # jump over the internal functions
    b(entry)

    label(func1)
    add(r0, 2)
    bx(lr)


            

Reported by Pylint.

              @micropython.asm_thumb
def f(r0):
    # jump over the internal functions
    b(entry)

    label(func1)
    add(r0, 2)
    bx(lr)


            

Reported by Pylint.

                  # jump over the internal functions
    b(entry)

    label(func1)
    add(r0, 2)
    bx(lr)

    label(func2)
    sub(r0, 1)

            

Reported by Pylint.

                  # jump over the internal functions
    b(entry)

    label(func1)
    add(r0, 2)
    bx(lr)

    label(func2)
    sub(r0, 1)

            

Reported by Pylint.

                  b(entry)

    label(func1)
    add(r0, 2)
    bx(lr)

    label(func2)
    sub(r0, 1)
    bx(lr)

            

Reported by Pylint.

              
    label(func1)
    add(r0, 2)
    bx(lr)

    label(func2)
    sub(r0, 1)
    bx(lr)


            

Reported by Pylint.

              
    label(func1)
    add(r0, 2)
    bx(lr)

    label(func2)
    sub(r0, 1)
    bx(lr)


            

Reported by Pylint.

                  add(r0, 2)
    bx(lr)

    label(func2)
    sub(r0, 1)
    bx(lr)

    label(entry)
    bl(func1)

            

Reported by Pylint.

                  add(r0, 2)
    bx(lr)

    label(func2)
    sub(r0, 1)
    bx(lr)

    label(entry)
    bl(func1)

            

Reported by Pylint.

              # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE

from board import LED
from machine import RTCounter, Temp
from ubluepy import Service, Characteristic, UUID, Peripheral, constants


def event_handler(id, handle, data):

            

Reported by Pylint.

              # THE SOFTWARE

from board import LED
from machine import RTCounter, Temp
from ubluepy import Service, Characteristic, UUID, Peripheral, constants


def event_handler(id, handle, data):
    global rtc

            

Reported by Pylint.

              
from board import LED
from machine import RTCounter, Temp
from ubluepy import Service, Characteristic, UUID, Peripheral, constants


def event_handler(id, handle, data):
    global rtc
    global periph

            

Reported by Pylint.

              from ubluepy import Service, Characteristic, UUID, Peripheral, constants


def event_handler(id, handle, data):
    global rtc
    global periph
    global serv_env_sense
    global notif_enabled


            

Reported by Pylint.

              from ubluepy import Service, Characteristic, UUID, Peripheral, constants


def event_handler(id, handle, data):
    global rtc
    global periph
    global serv_env_sense
    global notif_enabled


            

Reported by Pylint.

              

def event_handler(id, handle, data):
    global rtc
    global periph
    global serv_env_sense
    global notif_enabled

    if id == constants.EVT_GAP_CONNECTED:

            

Reported by Pylint.

              
def event_handler(id, handle, data):
    global rtc
    global periph
    global serv_env_sense
    global notif_enabled

    if id == constants.EVT_GAP_CONNECTED:
        # indicated 'connected'

            

Reported by Pylint.

              def event_handler(id, handle, data):
    global rtc
    global periph
    global serv_env_sense
    global notif_enabled

    if id == constants.EVT_GAP_CONNECTED:
        # indicated 'connected'
        LED(1).on()

            

Reported by Pylint.

                  global rtc
    global periph
    global serv_env_sense
    global notif_enabled

    if id == constants.EVT_GAP_CONNECTED:
        # indicated 'connected'
        LED(1).on()


            

Reported by Pylint.

                          rtc.stop()


def send_temp(timer_id):
    global notif_enabled
    global char_temp

    if notif_enabled:
        # measure chip temperature

            

Reported by Pylint.

              a.reinit()

# call __init__() after object is already init'd
class L(list):
    def reinit(self):
        super().__init__(range(2))
a = L(range(5))
print(a)
a.reinit()

            

Reported by Pylint.

              
# overriding list.__init__()
class L(list):
    def __init__(self, a, b):
        super().__init__([a, b])
print(L(2, 3))

# inherits implicitly from object
class A:

            

Reported by Pylint.

              # test subclassing a native type and overriding __init__

# overriding list.__init__()
class L(list):
    def __init__(self, a, b):
        super().__init__([a, b])
print(L(2, 3))

# inherits implicitly from object

            

Reported by Pylint.

              # test subclassing a native type and overriding __init__

# overriding list.__init__()
class L(list):
    def __init__(self, a, b):
        super().__init__([a, b])
print(L(2, 3))

# inherits implicitly from object

            

Reported by Pylint.

              # test subclassing a native type and overriding __init__

# overriding list.__init__()
class L(list):
    def __init__(self, a, b):
        super().__init__([a, b])
print(L(2, 3))

# inherits implicitly from object

            

Reported by Pylint.

              print(L(2, 3))

# inherits implicitly from object
class A:
    def __init__(self):
        print("A.__init__")
        super().__init__()
A()


            

Reported by Pylint.

              print(L(2, 3))

# inherits implicitly from object
class A:
    def __init__(self):
        print("A.__init__")
        super().__init__()
A()


            

Reported by Pylint.

              print(L(2, 3))

# inherits implicitly from object
class A:
    def __init__(self):
        print("A.__init__")
        super().__init__()
A()


            

Reported by Pylint.

              A()

# inherits explicitly from object
class B(object):
    def __init__(self):
        print("B.__init__")
        super().__init__()
B()


            

Reported by Pylint.

              A()

# inherits explicitly from object
class B(object):
    def __init__(self):
        print("B.__init__")
        super().__init__()
B()


            

Reported by Pylint.

              # Test BLE GAP connect/disconnect

from micropython import const
import time, machine, bluetooth

TIMEOUT_MS = 4000

_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)

            

Reported by Pylint.

              # Test BLE GAP connect/disconnect

from micropython import const
import time, machine, bluetooth

TIMEOUT_MS = 4000

_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)

            

Reported by Pylint.

              # Test BLE GAP connect/disconnect

from micropython import const
import time, machine, bluetooth

TIMEOUT_MS = 4000

_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)

            

Reported by Pylint.

              

def wait_for_event(event, timeout_ms):
    t0 = time.ticks_ms()
    while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
        if event in waiting_events:
            return waiting_events.pop(event)
        machine.idle()
    raise ValueError("Timeout waiting for {}".format(event))

            

Reported by Pylint.

              
def wait_for_event(event, timeout_ms):
    t0 = time.ticks_ms()
    while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
        if event in waiting_events:
            return waiting_events.pop(event)
        machine.idle()
    raise ValueError("Timeout waiting for {}".format(event))


            

Reported by Pylint.

              
def wait_for_event(event, timeout_ms):
    t0 = time.ticks_ms()
    while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
        if event in waiting_events:
            return waiting_events.pop(event)
        machine.idle()
    raise ValueError("Timeout waiting for {}".format(event))


            

Reported by Pylint.

              
# Acting in peripheral role.
def instance0():
    multitest.globals(BDADDR=ble.config("mac"))
    print("gap_advertise")
    ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY")
    multitest.next()
    try:
        # Wait for central to connect, then wait for it to disconnect.

            

Reported by Pylint.

                  multitest.globals(BDADDR=ble.config("mac"))
    print("gap_advertise")
    ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY")
    multitest.next()
    try:
        # Wait for central to connect, then wait for it to disconnect.
        wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
        wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS)


            

Reported by Pylint.

              
# Acting in central role.
def instance1():
    multitest.next()
    try:
        # Connect to peripheral and then disconnect.
        print("gap_connect")
        ble.gap_connect(*BDADDR)
        conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)

            

Reported by Pylint.

                  try:
        # Connect to peripheral and then disconnect.
        print("gap_connect")
        ble.gap_connect(*BDADDR)
        conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
        print("gap_disconnect:", ble.gap_disconnect(conn_handle))
        wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS)

        # Connect to peripheral and then let the peripheral disconnect us.

            

Reported by Pylint.

                      import asyncio
    except ImportError:
        print("SKIP")
        raise SystemExit


async def task(id, t):
    print("task start", id)
    await asyncio.sleep(t)

            

Reported by Pylint.

                      raise SystemExit


async def task(id, t):
    print("task start", id)
    await asyncio.sleep(t)
    print("task end", id)
    return id * 2


            

Reported by Pylint.

                  other.cancel()


async def task_wait_for_cancel(id, t, t_wait):
    print("task_wait_for_cancel start")
    try:
        await asyncio.wait_for(task(id, t), t_wait)
    except asyncio.CancelledError as er:
        print("task_wait_for_cancel cancelled")

            

Reported by Pylint.

              # Test asyncio.wait_for

try:
    import uasyncio as asyncio
except ImportError:
    try:
        import asyncio
    except ImportError:
        print("SKIP")

            

Reported by Pylint.

                      raise SystemExit


async def task(id, t):
    print("task start", id)
    await asyncio.sleep(t)
    print("task end", id)
    return id * 2


            

Reported by Pylint.

                      raise SystemExit


async def task(id, t):
    print("task start", id)
    await asyncio.sleep(t)
    print("task end", id)
    return id * 2


            

Reported by Pylint.

                      raise SystemExit


async def task(id, t):
    print("task start", id)
    await asyncio.sleep(t)
    print("task end", id)
    return id * 2


            

Reported by Pylint.

                  return id * 2


async def task_catch():
    print("task_catch start")
    try:
        await asyncio.sleep(0.2)
    except asyncio.CancelledError:
        print("ignore cancel")

            

Reported by Pylint.

                  print("task_catch done")


async def task_raise():
    print("task start")
    raise ValueError


async def task_cancel_other(t, other):

            

Reported by Pylint.

                  raise ValueError


async def task_cancel_other(t, other):
    print("task_cancel_other start")
    await asyncio.sleep(t)
    print("task_cancel_other cancel")
    other.cancel()


            

Reported by Pylint.

              print(f())

# multiple for loops within with
def f():
    with CtxMgr(1):
        for i in [1, 2]:
            for j in [3, 4]:
                return (i, j)
print(f())

            

Reported by Pylint.

              print(f())

# multiple for loops within nested withs
def f():
    with CtxMgr(1):
        for i in [1, 2]:
            for j in [3, 4]:
                with CtxMgr(2):
                    for k in [5, 6]:

            

Reported by Pylint.

              print(f())

# multiple for loops that are optimised, and nested withs
def f():
    with CtxMgr(1):
        for i in range(1, 3):
            for j in range(3, 5):
                with CtxMgr(2):
                    for k in range(5, 7):

            

Reported by Pylint.

              class CtxMgr:
    def __init__(self, id):
        self.id = id

    def __enter__(self):
        print("__enter__", self.id)
        return self

    def __exit__(self, a, b, c):

            

Reported by Pylint.

              class CtxMgr:
    def __init__(self, id):
        self.id = id

    def __enter__(self):
        print("__enter__", self.id)
        return self

    def __exit__(self, a, b, c):

            

Reported by Pylint.

              class CtxMgr:
    def __init__(self, id):
        self.id = id

    def __enter__(self):
        print("__enter__", self.id)
        return self

    def __exit__(self, a, b, c):

            

Reported by Pylint.

              class CtxMgr:
    def __init__(self, id):
        self.id = id

    def __enter__(self):
        print("__enter__", self.id)
        return self

    def __exit__(self, a, b, c):

            

Reported by Pylint.

                      print("__enter__", self.id)
        return self

    def __exit__(self, a, b, c):
        print("__exit__", self.id, repr(a), repr(b))

# simple case
def foo():
    with CtxMgr(1):

            

Reported by Pylint.

                      print("__enter__", self.id)
        return self

    def __exit__(self, a, b, c):
        print("__exit__", self.id, repr(a), repr(b))

# simple case
def foo():
    with CtxMgr(1):

            

Reported by Pylint.

                      print("__enter__", self.id)
        return self

    def __exit__(self, a, b, c):
        print("__exit__", self.id, repr(a), repr(b))

# simple case
def foo():
    with CtxMgr(1):

            

Reported by Pylint.

              # MicroPython uasyncio module
# MIT license; Copyright (c) 2019-2020 Damien P. George

from . import core


async def wait_for(aw, timeout, sleep=core.sleep):
    aw = core._promote_to_task(aw)
    if timeout is None:

            

Reported by Pylint.

              

async def wait_for(aw, timeout, sleep=core.sleep):
    aw = core._promote_to_task(aw)
    if timeout is None:
        return await aw

    def runner(waiter, aw):
        nonlocal status, result

            

Reported by Pylint.

                      try:
            result = await aw
            s = True
        except BaseException as er:
            s = er
        if status is None:
            # The waiter is still waiting, set status for it and cancel it.
            status = s
            waiter.cancel()

            

Reported by Pylint.

                          raise er
        else:
            # aw raised an exception, propagate it out to the caller.
            raise status

    # The sleep finished before aw, so cancel aw and raise TimeoutError.
    status = True
    runner_task.cancel()
    await runner_task

            

Reported by Pylint.

              

async def gather(*aws, return_exceptions=False):
    ts = [core._promote_to_task(aw) for aw in aws]
    for i in range(len(ts)):
        try:
            # TODO handle cancel of gather itself
            # if ts[i].coro:
            #    iter(ts[i]).waiting.push_head(cur_task)

            

Reported by Pylint.

                  ts = [core._promote_to_task(aw) for aw in aws]
    for i in range(len(ts)):
        try:
            # TODO handle cancel of gather itself
            # if ts[i].coro:
            #    iter(ts[i]).waiting.push_head(cur_task)
            #    try:
            #        yield
            #    except CancelledError as er:

            

Reported by Pylint.

                          #        # cancel all waiting tasks
            #        raise er
            ts[i] = await ts[i]
        except Exception as er:
            if return_exceptions:
                ts[i] = er
            else:
                raise er
    return ts

            

Reported by Pylint.

              # MicroPython uasyncio module
# MIT license; Copyright (c) 2019-2020 Damien P. George

from . import core


async def wait_for(aw, timeout, sleep=core.sleep):
    aw = core._promote_to_task(aw)
    if timeout is None:

            

Reported by Pylint.

              from . import core


async def wait_for(aw, timeout, sleep=core.sleep):
    aw = core._promote_to_task(aw)
    if timeout is None:
        return await aw

    def runner(waiter, aw):

            

Reported by Pylint.

              from . import core


async def wait_for(aw, timeout, sleep=core.sleep):
    aw = core._promote_to_task(aw)
    if timeout is None:
        return await aw

    def runner(waiter, aw):

            

Reported by Pylint.

                  compile
except NameError:
    print("SKIP")
    raise SystemExit

def test():
    global x

    c = compile("print(x)", "file", "exec")

            

Reported by Pylint.

                  raise SystemExit

def test():
    global x

    c = compile("print(x)", "file", "exec")

    try:
        exec(c)

            

Reported by Pylint.

                  c = compile("print(x)", "file", "exec")

    try:
        exec(c)
    except NameError:
        print("NameError")

    # global variable for compiled code to access
    x = 1

            

Reported by Bandit.

                  c = compile("print(x)", "file", "exec")

    try:
        exec(c)
    except NameError:
        print("NameError")

    # global variable for compiled code to access
    x = 1

            

Reported by Pylint.

                  # global variable for compiled code to access
    x = 1

    exec(c)

    exec(c, {"x":2})
    exec(c, {}, {"x":3})

    # single/eval mode

            

Reported by Pylint.

                  # global variable for compiled code to access
    x = 1

    exec(c)

    exec(c, {"x":2})
    exec(c, {}, {"x":3})

    # single/eval mode

            

Reported by Bandit.

              
    exec(c)

    exec(c, {"x":2})
    exec(c, {}, {"x":3})

    # single/eval mode
    exec(compile("if 1: 10 + 1\n", "file", "single"))
    exec(compile("print(10 + 2)", "file", "single"))

            

Reported by Bandit.

              
    exec(c)

    exec(c, {"x":2})
    exec(c, {}, {"x":3})

    # single/eval mode
    exec(compile("if 1: 10 + 1\n", "file", "single"))
    exec(compile("print(10 + 2)", "file", "single"))

            

Reported by Pylint.

                  exec(c)

    exec(c, {"x":2})
    exec(c, {}, {"x":3})

    # single/eval mode
    exec(compile("if 1: 10 + 1\n", "file", "single"))
    exec(compile("print(10 + 2)", "file", "single"))
    print(eval(compile("10 + 3", "file", "eval")))

            

Reported by Pylint.

                  exec(c)

    exec(c, {"x":2})
    exec(c, {}, {"x":3})

    # single/eval mode
    exec(compile("if 1: 10 + 1\n", "file", "single"))
    exec(compile("print(10 + 2)", "file", "single"))
    print(eval(compile("10 + 3", "file", "eval")))

            

Reported by Bandit.