foo(i)
    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 2
n_finished = 0

            

Reported by Pylint.

              import _thread


def foo(i):
    pass


def thread_entry(n, tup):
    for i in tup:

            

Reported by Pylint.

              import _thread


def foo(i):
    pass


def thread_entry(n, tup):
    for i in tup:

            

Reported by Pylint.

                  pass


def thread_entry(n, tup):
    for i in tup:
        foo(i)
    with lock:
        global n_finished
        n_finished += 1

            

Reported by Pylint.

                  pass


def thread_entry(n, tup):
    for i in tup:
        foo(i)
    with lock:
        global n_finished
        n_finished += 1

            

Reported by Pylint.

              

def thread_entry(n, tup):
    for i in tup:
        foo(i)
    with lock:
        global n_finished
        n_finished += 1


            

Reported by Pylint.

                  for i in tup:
        foo(i)
    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 2

            

Reported by Pylint.

              # test capability for threads to access a shared immutable data structure
#
# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd

import _thread


def foo(i):
    pass

            

Reported by Pylint.

              import _thread


def foo(i):
    pass


def thread_entry(n, tup):
    for i in tup:

            

Reported by Pylint.

              import _thread


def foo(i):
    pass


def thread_entry(n, tup):
    for i in tup:

            

Reported by Pylint.

                      foo(lst, idx)
    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 2
n_finished = 0

            

Reported by Pylint.

              import _thread


def foo(lst, i):
    lst[i] += 1


def thread_entry(n, lst, idx):
    for i in range(n):

            

Reported by Pylint.

              import _thread


def foo(lst, i):
    lst[i] += 1


def thread_entry(n, lst, idx):
    for i in range(n):

            

Reported by Pylint.

                  lst[i] += 1


def thread_entry(n, lst, idx):
    for i in range(n):
        foo(lst, idx)
    with lock:
        global n_finished
        n_finished += 1

            

Reported by Pylint.

              

def thread_entry(n, lst, idx):
    for i in range(n):
        foo(lst, idx)
    with lock:
        global n_finished
        n_finished += 1


            

Reported by Pylint.

              

def thread_entry(n, lst, idx):
    for i in range(n):
        foo(lst, idx)
    with lock:
        global n_finished
        n_finished += 1


            

Reported by Pylint.

                  for i in range(n):
        foo(lst, idx)
    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 2

            

Reported by Pylint.

              # test capability for threads to access a shared mutable data structure
# (without contention because they access different parts of the structure)
#
# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd

import _thread


def foo(lst, i):

            

Reported by Pylint.

              import _thread


def foo(lst, i):
    lst[i] += 1


def thread_entry(n, lst, idx):
    for i in range(n):

            

Reported by Pylint.

              import _thread


def foo(lst, i):
    lst[i] += 1


def thread_entry(n, lst, idx):
    for i in range(n):

            

Reported by Pylint.

                  sys.settrace
except AttributeError:
    print("SKIP")
    raise SystemExit


def print_stacktrace(frame, level=0):
    print(
        "%2d: %s@%s:%s => %s:%d"

            

Reported by Pylint.

              trace_count = 0


def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)
    trace_count += 1
    print_stacktrace(frame)
    return trace_tick_handler

            

Reported by Pylint.

              

def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)
    trace_count += 1
    print_stacktrace(frame)
    return trace_tick_handler


            

Reported by Pylint.

              # test sys.settrace with generators

import sys

try:
    sys.settrace
except AttributeError:
    print("SKIP")
    raise SystemExit

            

Reported by Pylint.

                  raise SystemExit


def print_stacktrace(frame, level=0):
    print(
        "%2d: %s@%s:%s => %s:%d"
        % (
            level,
            "  ",

            

Reported by Pylint.

                      print_stacktrace(frame.f_back, level + 1)


trace_count = 0


def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)

            

Reported by Pylint.

              trace_count = 0


def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)
    trace_count += 1
    print_stacktrace(frame)
    return trace_tick_handler

            

Reported by Pylint.

              

def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)
    trace_count += 1
    print_stacktrace(frame)
    return trace_tick_handler


            

Reported by Pylint.

                  return trace_tick_handler


def test_generator():
    def make_gen():
        yield 1 << 0
        yield 1 << 1
        yield 1 << 2
        return 1 << 3

            

Reported by Pylint.

                      return 1 << 3

    gen = make_gen()
    r = 0
    try:

        r += gen.send(None)

        while True:

            

Reported by Pylint.

              except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")
    raise SystemExit



            

Reported by Pylint.

              # test that modtls produces a numerical error message when out of heap

try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:

            

Reported by Pylint.

              
try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")

            

Reported by Pylint.

                  from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")
    raise SystemExit


# test with heap locked to see it switch to number-only error message
def test(addr):
    alloc_emergency_exception_buf(256)

            

Reported by Pylint.

              

# test with heap locked to see it switch to number-only error message
def test(addr):
    alloc_emergency_exception_buf(256)
    s = socket.socket()
    s.connect(addr)
    try:
        s.setblocking(False)

            

Reported by Pylint.

                  s.connect(addr)
    try:
        s.setblocking(False)
        s = ssl.wrap_socket(s, do_handshake=False)
        heap_lock()
        print("heap is locked")
        while True:
            ret = s.write("foo")
            if ret:

            

Reported by Pylint.

              # test that modtls produces a numerical error message when out of heap

try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:

            

Reported by Pylint.

              # test that modtls produces a numerical error message when out of heap

try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")
    raise SystemExit

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")
    raise SystemExit

            

Reported by Pylint.

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


async def task(s, allow_cancel):
    try:
        print("task start")

            

Reported by Pylint.

              # Test cancelling a task

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

            

Reported by Pylint.

                      raise SystemExit


async def task(s, allow_cancel):
    try:
        print("task start")
        await asyncio.sleep(s)
        print("task done")
    except asyncio.CancelledError as er:

            

Reported by Pylint.

                      raise SystemExit


async def task(s, allow_cancel):
    try:
        print("task start")
        await asyncio.sleep(s)
        print("task done")
    except asyncio.CancelledError as er:

            

Reported by Pylint.

                      print("task start")
        await asyncio.sleep(s)
        print("task done")
    except asyncio.CancelledError as er:
        print("task cancel")
        if allow_cancel:
            raise er



            

Reported by Pylint.

                          raise er


async def task2(allow_cancel):
    print("task 2")
    try:
        await asyncio.create_task(task(0.05, allow_cancel))
    except asyncio.CancelledError as er:
        print("task 2 cancel")

            

Reported by Pylint.

                  print("task 2")
    try:
        await asyncio.create_task(task(0.05, allow_cancel))
    except asyncio.CancelledError as er:
        print("task 2 cancel")
        raise er
    print("task 2 done")



            

Reported by Pylint.

                  print("task 2 done")


async def main():
    # Cancel task immediately
    t = asyncio.create_task(task(2, True))
    print(t.cancel())

    # Cancel task after it has started

            

Reported by Pylint.

              
async def main():
    # Cancel task immediately
    t = asyncio.create_task(task(2, True))
    print(t.cancel())

    # Cancel task after it has started
    t = asyncio.create_task(task(2, True))
    await asyncio.sleep(0.01)

            

Reported by Pylint.

                  print(t.cancel())

    # Cancel task after it has started
    t = asyncio.create_task(task(2, True))
    await asyncio.sleep(0.01)
    print(t.cancel())
    print("main sleep")
    await asyncio.sleep(0.01)


            

Reported by Pylint.

              

class PortData:
    def __init__(self, name, dir, output, make_flags=None):
        self.name = name
        self.dir = dir
        self.output = output
        self.make_flags = make_flags
        self.needs_mpy_cross = dir not in ("bare-arm", "minimal")

            

Reported by Pylint.

              
"""

import collections, sys, re, subprocess

MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]


class PortData:

            

Reported by Bandit.

              
"""

import collections, sys, re, subprocess

MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]


class PortData:

            

Reported by Pylint.

              MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]


class PortData:
    def __init__(self, name, dir, output, make_flags=None):
        self.name = name
        self.dir = dir
        self.output = output
        self.make_flags = make_flags

            

Reported by Pylint.

              MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]


class PortData:
    def __init__(self, name, dir, output, make_flags=None):
        self.name = name
        self.dir = dir
        self.output = output
        self.make_flags = make_flags

            

Reported by Pylint.

              }


def syscmd(*args):
    sys.stdout.flush()
    a2 = []
    for a in args:
        if isinstance(a, str):
            a2.append(a)

            

Reported by Pylint.

              
def syscmd(*args):
    sys.stdout.flush()
    a2 = []
    for a in args:
        if isinstance(a, str):
            a2.append(a)
        elif a:
            a2.extend(a)

            

Reported by Pylint.

              def syscmd(*args):
    sys.stdout.flush()
    a2 = []
    for a in args:
        if isinstance(a, str):
            a2.append(a)
        elif a:
            a2.extend(a)
    subprocess.check_call(a2)

            

Reported by Pylint.

                          a2.append(a)
        elif a:
            a2.extend(a)
    subprocess.check_call(a2)


def parse_port_list(args):
    if not args:
        return list(port_data.values())

            

Reported by Bandit.

                  subprocess.check_call(a2)


def parse_port_list(args):
    if not args:
        return list(port_data.values())
    else:
        ports = []
        for arg in args:

            

Reported by Pylint.

              # This example demonstrates a peripheral implementing the Nordic UART Service (NUS).

import bluetooth
from ble_advertising import advertising_payload

from micropython import const

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

            

Reported by Pylint.

              import bluetooth
from ble_advertising import advertising_payload

from micropython import const

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


            

Reported by Pylint.

                      while True:
            uart.write(str(nums[i]) + "\n")
            i = (i + 1) % len(nums)
            time.sleep_ms(1000)
    except KeyboardInterrupt:
        pass

    uart.close()


            

Reported by Pylint.

              # This example demonstrates a peripheral implementing the Nordic UART Service (NUS).

import bluetooth
from ble_advertising import advertising_payload

from micropython import const

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

            

Reported by Pylint.

              import bluetooth
from ble_advertising import advertising_payload

from micropython import const

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


            

Reported by Pylint.

              _ADV_APPEARANCE_GENERIC_COMPUTER = const(128)


class BLEUART:
    def __init__(self, ble, name="mpy-uart", rxbuf=100):
        self._ble = ble
        self._ble.active(True)
        self._ble.irq(self._irq)
        ((self._tx_handle, self._rx_handle),) = self._ble.gatts_register_services((_UART_SERVICE,))

            

Reported by Pylint.

                      self._payload = advertising_payload(name=name, appearance=_ADV_APPEARANCE_GENERIC_COMPUTER)
        self._advertise()

    def irq(self, handler):
        self._handler = handler

    def _irq(self, event, data):
        # Track connections so we can send notifications.
        if event == _IRQ_CENTRAL_CONNECT:

            

Reported by Pylint.

                              if self._handler:
                    self._handler()

    def any(self):
        return len(self._rx_buffer)

    def read(self, sz=None):
        if not sz:
            sz = len(self._rx_buffer)

            

Reported by Pylint.

                  def any(self):
        return len(self._rx_buffer)

    def read(self, sz=None):
        if not sz:
            sz = len(self._rx_buffer)
        result = self._rx_buffer[0:sz]
        self._rx_buffer = self._rx_buffer[sz:]
        return result

            

Reported by Pylint.

                  def any(self):
        return len(self._rx_buffer)

    def read(self, sz=None):
        if not sz:
            sz = len(self._rx_buffer)
        result = self._rx_buffer[0:sz]
        self._rx_buffer = self._rx_buffer[sz:]
        return result

            

Reported by Pylint.

              

# basic with
@micropython.viper
def f():
    with C():
        print(1)



            

Reported by Pylint.

              f()

# nested with and try-except
@micropython.viper
def f():
    try:
        with C():
            print(1)
            fail

            

Reported by Pylint.

              
# nested with and try-except
@micropython.viper
def f():
    try:
        with C():
            print(1)
            fail
            print(2)

            

Reported by Pylint.

                  try:
        with C():
            print(1)
            fail
            print(2)
    except NameError:
        print("NameError")



            

Reported by Pylint.

              # test with handling within a viper function


class C:
    def __init__(self):
        print("__init__")

    def __enter__(self):
        print("__enter__")

            

Reported by Pylint.

              # test with handling within a viper function


class C:
    def __init__(self):
        print("__init__")

    def __enter__(self):
        print("__enter__")

            

Reported by Pylint.

              # test with handling within a viper function


class C:
    def __init__(self):
        print("__init__")

    def __enter__(self):
        print("__enter__")

            

Reported by Pylint.

                  def __enter__(self):
        print("__enter__")

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


# basic with
@micropython.viper

            

Reported by Pylint.

                  def __enter__(self):
        print("__enter__")

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


# basic with
@micropython.viper

            

Reported by Pylint.

                  def __enter__(self):
        print("__enter__")

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


# basic with
@micropython.viper

            

Reported by Pylint.

                  with lock:
        print(sum, lst[-1])
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 10
n_finished = 0

            

Reported by Pylint.

              
    # run a loop which allocates a small list and uses it each iteration
    lst = 8 * [0]
    sum = 0
    for i in range(n):
        sum += last(lst)
        lst = [0, 0, 0, 0, 0, 0, 0, i + 1]

    # check that the bytearray still has the right data

            

Reported by Pylint.

                  # run a loop which allocates a small list and uses it each iteration
    lst = 8 * [0]
    sum = 0
    for i in range(n):
        sum += last(lst)
        lst = [0, 0, 0, 0, 0, 0, 0, i + 1]

    # check that the bytearray still has the right data
    for i, b in enumerate(data):

            

Reported by Pylint.

                  # print the result of the loop and indicate we are finished
    with lock:
        print(sum, lst[-1])
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 10

            

Reported by Pylint.

              # stress test for the heap by allocating lots of objects within threads
# allocates about 5mb on the heap
#
# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd

try:
    import utime as time
except ImportError:
    import time

            

Reported by Pylint.

              import _thread


def last(l):
    return l[-1]


def thread_entry(n):
    # allocate a bytearray and fill it

            

Reported by Pylint.

              import _thread


def last(l):
    return l[-1]


def thread_entry(n):
    # allocate a bytearray and fill it

            

Reported by Pylint.

                  return l[-1]


def thread_entry(n):
    # allocate a bytearray and fill it
    data = bytearray(i for i in range(256))

    # run a loop which allocates a small list and uses it each iteration
    lst = 8 * [0]

            

Reported by Pylint.

                  return l[-1]


def thread_entry(n):
    # allocate a bytearray and fill it
    data = bytearray(i for i in range(256))

    # run a loop which allocates a small list and uses it each iteration
    lst = 8 * [0]

            

Reported by Pylint.

                      lst = [0, 0, 0, 0, 0, 0, 0, i + 1]

    # check that the bytearray still has the right data
    for i, b in enumerate(data):
        assert i == b

    # print the result of the loop and indicate we are finished
    with lock:
        print(sum, lst[-1])

            

Reported by Pylint.

                  print("try1")
    try:
        print("try2")
        foo()
    except:
        print("except2")
finally:
    print("finally1")
print()

            

Reported by Pylint.

                      print("try1")
        try:
            print("try2")
            foo()
        finally:
            print("finally2")
    finally:
        print("finally1")
except:

            

Reported by Pylint.

              finally:
    try:
        print("try2")
        foo
    except:
        print("except2")
    print("finally1")
print()


            

Reported by Pylint.

                      print("try")
    finally:
        print("finally")
    foo
try:
    func()
except:
    print("except")

            

Reported by Pylint.

                  try:
        print("try2")
        foo()
    except:
        print("except2")
finally:
    print("finally1")
print()


            

Reported by Pylint.

                          print("finally2")
    finally:
        print("finally1")
except:
    print("catch-all except")
print()

# case where a try-except within a finally cancels the exception
print("exc-finally-subexcept")

            

Reported by Pylint.

              finally:
    try:
        print("try2")
        foo
    except:
        print("except2")
    print("finally1")
print()


            

Reported by Pylint.

                  try:
        print("try2")
        foo
    except:
        print("except2")
    print("finally1")
print()

# case where exception is raised after a finally has finished (tests that the finally doesn't run again)

            

Reported by Pylint.

                      print("try")
    finally:
        print("finally")
    foo
try:
    func()
except:
    print("except")

            

Reported by Pylint.

                  foo
try:
    func()
except:
    print("except")

            

Reported by Pylint.