ev = asyncio.Event()
    server = await asyncio.start_server(handle_connection, "0.0.0.0", PORT)
    print("server running")
    multitest.next()
    async with server:
        await asyncio.wait_for(ev.wait(), 10)


async def tcp_client():

            

Reported by Pylint.

              

async def tcp_client():
    reader, writer = await asyncio.open_connection(IP, PORT)
    print(await reader.readexactly(2))
    print(await reader.readexactly(0))
    print(await reader.readexactly(1))
    try:
        print(await reader.readexactly(2))

            

Reported by Pylint.

              

def instance0():
    multitest.globals(IP=multitest.get_network_ip())
    asyncio.run(tcp_server())


def instance1():
    multitest.next()

            

Reported by Pylint.

              

def instance0():
    multitest.globals(IP=multitest.get_network_ip())
    asyncio.run(tcp_server())


def instance1():
    multitest.next()

            

Reported by Pylint.

              

def instance1():
    multitest.next()
    asyncio.run(tcp_client())

            

Reported by Pylint.

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

PORT = 8000


async def handle_connection(reader, writer):

            

Reported by Pylint.

              PORT = 8000


async def handle_connection(reader, writer):
    writer.write(b"a")
    await writer.drain()

    # Split the first 2 bytes up so the client must wait for the second one
    await asyncio.sleep(0.1)

            

Reported by Pylint.

              

async def tcp_server():
    global ev
    ev = asyncio.Event()
    server = await asyncio.start_server(handle_connection, "0.0.0.0", PORT)
    print("server running")
    multitest.next()
    async with server:

            

Reported by Pylint.

              async def tcp_server():
    global ev
    ev = asyncio.Event()
    server = await asyncio.start_server(handle_connection, "0.0.0.0", PORT)
    print("server running")
    multitest.next()
    async with server:
        await asyncio.wait_for(ev.wait(), 10)


            

Reported by Bandit.

              

async def tcp_client():
    reader, writer = await asyncio.open_connection(IP, PORT)
    print(await reader.readexactly(2))
    print(await reader.readexactly(0))
    print(await reader.readexactly(1))
    try:
        print(await reader.readexactly(2))

            

Reported by Pylint.

              try:
    import uos

    uos.VfsFat
    uos.VfsLfs2
except (ImportError, AttributeError):
    print("SKIP")
    raise SystemExit


            

Reported by Pylint.

                  import uos

    uos.VfsFat
    uos.VfsLfs2
except (ImportError, AttributeError):
    print("SKIP")
    raise SystemExit



            

Reported by Pylint.

                  uos.VfsLfs2
except (ImportError, AttributeError):
    print("SKIP")
    raise SystemExit


class RAMBlockDevice:
    ERASE_BLOCK_SIZE = 512


            

Reported by Pylint.

                      for i in range(len(buf)):
            self.data[addr + i] = buf[i]

    def ioctl(self, op, arg):
        if op == 4:  # block count
            return len(self.data) // self.ERASE_BLOCK_SIZE
        if op == 5:  # block size
            return self.ERASE_BLOCK_SIZE
        if op == 6:  # erase block

            

Reported by Pylint.

                          return 0


def test(bdev, vfs_class):
    print("test", vfs_class)

    # mkfs
    vfs_class.mkfs(bdev)


            

Reported by Pylint.

              
    # open, write close
    f = vfs.open("test", "w")
    for i in range(10):
        f.write("some data")
    f.close()

    # ilistdir
    print(list(vfs.ilistdir()))

            

Reported by Pylint.

                  bdev = RAMBlockDevice(50)
except MemoryError:
    print("SKIP")
    raise SystemExit

test(bdev, uos.VfsFat)
test(bdev, uos.VfsLfs2)

            

Reported by Pylint.

              # Test for behaviour of combined standard and extended block device

try:
    import uos

    uos.VfsFat
    uos.VfsLfs2
except (ImportError, AttributeError):
    print("SKIP")

            

Reported by Pylint.

                  raise SystemExit


class RAMBlockDevice:
    ERASE_BLOCK_SIZE = 512

    def __init__(self, blocks):
        self.data = bytearray(blocks * self.ERASE_BLOCK_SIZE)


            

Reported by Pylint.

                  def __init__(self, blocks):
        self.data = bytearray(blocks * self.ERASE_BLOCK_SIZE)

    def readblocks(self, block, buf, off=0):
        addr = block * self.ERASE_BLOCK_SIZE + off
        for i in range(len(buf)):
            buf[i] = self.data[addr + i]

    def writeblocks(self, block, buf, off=None):

            

Reported by Pylint.

              
    with lock:
        global n_finished
        n_finished += 1


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

            

Reported by Pylint.

              
# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            l = len(ba)
            ba.append(i)
            assert len(ba) >= l + 1


            

Reported by Pylint.

              # main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            l = len(ba)
            ba.append(i)
            assert len(ba) >= l + 1

            l = len(ba)

            

Reported by Pylint.

                          assert len(ba) >= l + 1

    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 4

            

Reported by Pylint.

              # test concurrent mutating access to a shared bytearray object
#
# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd

import _thread

# the shared bytearray
ba = bytearray()


            

Reported by Pylint.

              ba = bytearray()

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            l = len(ba)
            ba.append(i)
            assert len(ba) >= l + 1

            

Reported by Pylint.

              ba = bytearray()

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            l = len(ba)
            ba.append(i)
            assert len(ba) >= l + 1

            

Reported by Pylint.

              ba = bytearray()

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            l = len(ba)
            ba.append(i)
            assert len(ba) >= l + 1

            

Reported by Pylint.

              ba = bytearray()

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            l = len(ba)
            ba.append(i)
            assert len(ba) >= l + 1

            

Reported by Pylint.

              ba = bytearray()

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            l = len(ba)
            ba.append(i)
            assert len(ba) >= l + 1

            

Reported by Pylint.

                  print("task done")


async def main():
    print("main start")
    t = asyncio.create_task(task())
    await t
    print("main done")


            

Reported by Pylint.

              
# main task raising an exception with sub-task not yet scheduled
# TODO not currently working, task is never scheduled
async def task():
    # print('task run') uncomment this line when it works
    pass


async def main():

            

Reported by Pylint.

                  pass


async def main():
    print("main start")
    asyncio.create_task(task())
    raise ValueError(3)
    print("main done")


            

Reported by Pylint.

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

# main task raising an exception
async def main():
    print("main start")
    raise ValueError(1)

            

Reported by Pylint.

              async def main():
    print("main start")
    raise ValueError(1)
    print("main done")


try:
    asyncio.run(main())
except ValueError as er:

            

Reported by Pylint.

              async def task():
    print("task start")
    raise ValueError(2)
    print("task done")


async def main():
    print("main start")
    t = asyncio.create_task(task())

            

Reported by Pylint.

                  print("ValueError", er.args[0])

# main task raising an exception with sub-task not yet scheduled
# TODO not currently working, task is never scheduled
async def task():
    # print('task run') uncomment this line when it works
    pass



            

Reported by Pylint.

                  print("main start")
    asyncio.create_task(task())
    raise ValueError(3)
    print("main done")


try:
    asyncio.run(main())
except ValueError as er:

            

Reported by Pylint.

              # Test general exception handling

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

            

Reported by Pylint.

                      raise SystemExit

# main task raising an exception
async def main():
    print("main start")
    raise ValueError(1)
    print("main done")



            

Reported by Pylint.

              RTC test for the CC3200 based boards.
"""

from machine import RTC
import os
import time

mch = os.uname().machine
if not "LaunchPad" in mch and not "WiPy" in mch:

            

Reported by Pylint.

              import os
import time

mch = os.uname().machine
if not "LaunchPad" in mch and not "WiPy" in mch:
    raise Exception("Board not supported!")

rtc = RTC()
print(rtc)

            

Reported by Pylint.

              rtc.init((2015, 8, 29, 9, 0, 0, 0, None))
print(rtc.now()[:6])
seconds = rtc.now()[5]
time.sleep_ms(1000)
print(rtc.now()[5] - seconds == 1)
seconds = rtc.now()[5]
time.sleep_ms(2000)
print(rtc.now()[5] - seconds == 2)


            

Reported by Pylint.

              time.sleep_ms(1000)
print(rtc.now()[5] - seconds == 1)
seconds = rtc.now()[5]
time.sleep_ms(2000)
print(rtc.now()[5] - seconds == 2)

# initialization with shorter tuples
rtc.init((2015, 9, 19, 8, 0, 0, 0))
print(rtc.now()[5])

            

Reported by Pylint.

              rtc.init((2015, 8, 29, 9, 0, 0, 0, None))
rtc.alarm(0, 5000)
rtc.alarm(time=2000)
time.sleep_ms(1000)
left = rtc.alarm_left()
print(abs(left - 1000) <= 10)
time.sleep_ms(1000)
print(rtc.alarm_left() == 0)
time.sleep_ms(100)

            

Reported by Pylint.

              time.sleep_ms(1000)
left = rtc.alarm_left()
print(abs(left - 1000) <= 10)
time.sleep_ms(1000)
print(rtc.alarm_left() == 0)
time.sleep_ms(100)
print(rtc.alarm_left(0) == 0)

rtc.alarm(time=1000, repeat=True)

            

Reported by Pylint.

              print(abs(left - 1000) <= 10)
time.sleep_ms(1000)
print(rtc.alarm_left() == 0)
time.sleep_ms(100)
print(rtc.alarm_left(0) == 0)

rtc.alarm(time=1000, repeat=True)
time.sleep_ms(1500)
left = rtc.alarm_left()

            

Reported by Pylint.

              print(rtc.alarm_left(0) == 0)

rtc.alarm(time=1000, repeat=True)
time.sleep_ms(1500)
left = rtc.alarm_left()
print(abs(left - 500) <= 15)

rtc.init((2015, 8, 29, 9, 0, 0, 0, None))
rtc.alarm(time=(2015, 8, 29, 9, 0, 45))

            

Reported by Pylint.

              
rtc.init((2015, 8, 29, 9, 0, 0, 0, None))
rtc.alarm(time=(2015, 8, 29, 9, 0, 45))
time.sleep_ms(1000)
left = rtc.alarm_left()
print(abs(left - 44000) <= 90)
rtc.alarm_cancel()
rtc.deinit()


            

Reported by Pylint.

              # next ones must raise
try:
    rtc.alarm(5000)
except:
    print("Exception")

try:
    rtc.alarm_left(1)
except:

            

Reported by Pylint.

              def get_version_info_from_git():
    # Python 2.6 doesn't have check_output, so check for that
    try:
        subprocess.check_output
        subprocess.check_call
    except AttributeError:
        return None

    # Note: git describe doesn't work if no tag is available

            

Reported by Pylint.

                  # Python 2.6 doesn't have check_output, so check for that
    try:
        subprocess.check_output
        subprocess.check_call
    except AttributeError:
        return None

    # Note: git describe doesn't work if no tag is available
    try:

            

Reported by Pylint.

              import sys
import os
import datetime
import subprocess


def get_version_info_from_git():
    # Python 2.6 doesn't have check_output, so check for that
    try:

            

Reported by Bandit.

              import subprocess


def get_version_info_from_git():
    # Python 2.6 doesn't have check_output, so check for that
    try:
        subprocess.check_output
        subprocess.check_call
    except AttributeError:

            

Reported by Pylint.

              
    # Note: git describe doesn't work if no tag is available
    try:
        git_tag = subprocess.check_output(
            ["git", "describe", "--tags", "--dirty", "--always", "--match", "v[1-9].*"],
            stderr=subprocess.STDOUT,
            universal_newlines=True,
        ).strip()
    except subprocess.CalledProcessError as er:

            

Reported by Bandit.

              
    # Note: git describe doesn't work if no tag is available
    try:
        git_tag = subprocess.check_output(
            ["git", "describe", "--tags", "--dirty", "--always", "--match", "v[1-9].*"],
            stderr=subprocess.STDOUT,
            universal_newlines=True,
        ).strip()
    except subprocess.CalledProcessError as er:

            

Reported by Bandit.

                          stderr=subprocess.STDOUT,
            universal_newlines=True,
        ).strip()
    except subprocess.CalledProcessError as er:
        if er.returncode == 128:
            # git exit code of 128 means no repository found
            return None
        git_tag = ""
    except OSError:

            

Reported by Pylint.

                  except OSError:
        return None
    try:
        git_hash = subprocess.check_output(
            ["git", "rev-parse", "--short", "HEAD"],
            stderr=subprocess.STDOUT,
            universal_newlines=True,
        ).strip()
    except subprocess.CalledProcessError:

            

Reported by Bandit.

                  except OSError:
        return None
    try:
        git_hash = subprocess.check_output(
            ["git", "rev-parse", "--short", "HEAD"],
            stderr=subprocess.STDOUT,
            universal_newlines=True,
        ).strip()
    except subprocess.CalledProcessError:

            

Reported by Bandit.

              
    try:
        # Check if there are any modified files.
        subprocess.check_call(
            ["git", "diff", "--no-ext-diff", "--quiet", "--exit-code"], stderr=subprocess.STDOUT
        )
        # Check if there are any staged files.
        subprocess.check_call(
            ["git", "diff-index", "--cached", "--quiet", "HEAD", "--"], stderr=subprocess.STDOUT

            

Reported by Bandit.

              # unsupported unary ops

try:
    ~1.2
except TypeError:
    print("TypeError")

try:
    1.2 in 3.4

            

Reported by Pylint.

                  print("TypeError")

try:
    1.2 in 3.4
except TypeError:
    print("TypeError")

# small int on LHS, float on RHS, unsupported op
try:

            

Reported by Pylint.

              # test basic float capabilities

# literals
print(0.12)
print(1.0)
print(1.2)
print(0e0)
print(0e0)
print(0e-0)

            

Reported by Pylint.

              print(-(1.2))

# division of integers
x = 1 / 2
print(x)

# /= operator
a = 1
a /= 2

            

Reported by Pylint.

              print(x)

# /= operator
a = 1
a /= 2
print(a)

# floor division
print(1.0 // 2)

            

Reported by Pylint.

              
# comparison
print(1.2 <= 3.4)
print(1.2 <= -3.4)
print(1.2 >= 3.4)
print(1.2 >= -3.4)
print(0.0 == False, 1.0 == True)
print(False == 0.0, True == 1.0)


            

Reported by Pylint.

              print(1.2 <= 3.4)
print(1.2 <= -3.4)
print(1.2 >= 3.4)
print(1.2 >= -3.4)
print(0.0 == False, 1.0 == True)
print(False == 0.0, True == 1.0)

# comparison of nan is special
nan = float("nan")

            

Reported by Pylint.

              print(1.2 <= -3.4)
print(1.2 >= 3.4)
print(1.2 >= -3.4)
print(0.0 == False, 1.0 == True)
print(False == 0.0, True == 1.0)

# comparison of nan is special
nan = float("nan")
print(nan == 1.2)

            

Reported by Pylint.

              print(1.2 <= -3.4)
print(1.2 >= 3.4)
print(1.2 >= -3.4)
print(0.0 == False, 1.0 == True)
print(False == 0.0, True == 1.0)

# comparison of nan is special
nan = float("nan")
print(nan == 1.2)

            

Reported by Pylint.

              print(1.2 <= -3.4)
print(1.2 >= 3.4)
print(1.2 >= -3.4)
print(0.0 == False, 1.0 == True)
print(False == 0.0, True == 1.0)

# comparison of nan is special
nan = float("nan")
print(nan == 1.2)

            

Reported by Pylint.

              # test for type.__bases__ implementation

if not hasattr(object, '__bases__'):
    print("SKIP")
    raise SystemExit

class A:
    pass


            

Reported by Pylint.

                  print("SKIP")
    raise SystemExit

class A:
    pass

class B(object):
    pass


            

Reported by Pylint.

                  print("SKIP")
    raise SystemExit

class A:
    pass

class B(object):
    pass


            

Reported by Pylint.

                  print("SKIP")
    raise SystemExit

class A:
    pass

class B(object):
    pass


            

Reported by Pylint.

              class A:
    pass

class B(object):
    pass

class C(B):
    pass


            

Reported by Pylint.

              class A:
    pass

class B(object):
    pass

class C(B):
    pass


            

Reported by Pylint.

              class A:
    pass

class B(object):
    pass

class C(B):
    pass


            

Reported by Pylint.

              class A:
    pass

class B(object):
    pass

class C(B):
    pass


            

Reported by Pylint.

              class B(object):
    pass

class C(B):
    pass

class D(C, A):
    pass


            

Reported by Pylint.

              class B(object):
    pass

class C(B):
    pass

class D(C, A):
    pass


            

Reported by Pylint.

                  global ev
    ev = asyncio.Event()
    server = await asyncio.start_server(handle_connection, "0.0.0.0", PORT)
    multitest.next()
    async with server:
        await asyncio.wait_for(ev.wait(), 10)


def instance0():

            

Reported by Pylint.

              

def instance0():
    multitest.globals(IP=multitest.get_network_ip())
    asyncio.run(main())


def instance1():
    if not hasattr(socket, "SO_LINGER"):

            

Reported by Pylint.

              

def instance0():
    multitest.globals(IP=multitest.get_network_ip())
    asyncio.run(main())


def instance1():
    if not hasattr(socket, "SO_LINGER"):

            

Reported by Pylint.

              
def instance1():
    if not hasattr(socket, "SO_LINGER"):
        multitest.skip()
    multitest.next()
    s = socket.socket()
    s.connect(socket.getaddrinfo(IP, PORT)[0][-1])
    lgr_onoff = 1
    lgr_linger = 0

            

Reported by Pylint.

              def instance1():
    if not hasattr(socket, "SO_LINGER"):
        multitest.skip()
    multitest.next()
    s = socket.socket()
    s.connect(socket.getaddrinfo(IP, PORT)[0][-1])
    lgr_onoff = 1
    lgr_linger = 0
    s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", lgr_onoff, lgr_linger))

            

Reported by Pylint.

                      multitest.skip()
    multitest.next()
    s = socket.socket()
    s.connect(socket.getaddrinfo(IP, PORT)[0][-1])
    lgr_onoff = 1
    lgr_linger = 0
    s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", lgr_onoff, lgr_linger))
    s.send(b"GET / HTTP/1.0\r\n\r\n")
    time.sleep(0.1)

            

Reported by Pylint.

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

import struct, time, socket

PORT = 8000


            

Reported by Pylint.

              

async def main():
    global ev
    ev = asyncio.Event()
    server = await asyncio.start_server(handle_connection, "0.0.0.0", PORT)
    multitest.next()
    async with server:
        await asyncio.wait_for(ev.wait(), 10)

            

Reported by Pylint.

              async def main():
    global ev
    ev = asyncio.Event()
    server = await asyncio.start_server(handle_connection, "0.0.0.0", PORT)
    multitest.next()
    async with server:
        await asyncio.wait_for(ev.wait(), 10)



            

Reported by Bandit.

              # Test TCP server with client issuing TCP RST part way through read

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

            

Reported by Pylint.

              try:
    import uos

    uos.VfsLfs1
    uos.VfsLfs2
except (ImportError, AttributeError):
    print("SKIP")
    raise SystemExit


            

Reported by Pylint.

                  import uos

    uos.VfsLfs1
    uos.VfsLfs2
except (ImportError, AttributeError):
    print("SKIP")
    raise SystemExit



            

Reported by Pylint.

                  uos.VfsLfs2
except (ImportError, AttributeError):
    print("SKIP")
    raise SystemExit


class RAMBlockDevice:
    ERASE_BLOCK_SIZE = 1024


            

Reported by Pylint.

                      for i in range(len(buf)):
            self.data[addr + i] = buf[i]

    def ioctl(self, op, arg):
        if op == 4:  # block count
            return len(self.data) // self.ERASE_BLOCK_SIZE
        if op == 5:  # block size
            return self.ERASE_BLOCK_SIZE
        if op == 6:  # erase block

            

Reported by Pylint.

                          return 0


def test(bdev, vfs_class):
    print("test", vfs_class)

    # mkfs with too-small block device
    try:
        vfs_class.mkfs(RAMBlockDevice(1))

            

Reported by Pylint.

              # Test for VfsLittle using a RAM device, testing error handling

try:
    import uos

    uos.VfsLfs1
    uos.VfsLfs2
except (ImportError, AttributeError):
    print("SKIP")

            

Reported by Pylint.

                  raise SystemExit


class RAMBlockDevice:
    ERASE_BLOCK_SIZE = 1024

    def __init__(self, blocks):
        self.data = bytearray(blocks * self.ERASE_BLOCK_SIZE)


            

Reported by Pylint.

                  def __init__(self, blocks):
        self.data = bytearray(blocks * self.ERASE_BLOCK_SIZE)

    def readblocks(self, block, buf, off):
        addr = block * self.ERASE_BLOCK_SIZE + off
        for i in range(len(buf)):
            buf[i] = self.data[addr + i]

    def writeblocks(self, block, buf, off):

            

Reported by Pylint.

              
    def readblocks(self, block, buf, off):
        addr = block * self.ERASE_BLOCK_SIZE + off
        for i in range(len(buf)):
            buf[i] = self.data[addr + i]

    def writeblocks(self, block, buf, off):
        addr = block * self.ERASE_BLOCK_SIZE + off
        for i in range(len(buf)):

            

Reported by Pylint.

                      for i in range(len(buf)):
            buf[i] = self.data[addr + i]

    def writeblocks(self, block, buf, off):
        addr = block * self.ERASE_BLOCK_SIZE + off
        for i in range(len(buf)):
            self.data[addr + i] = buf[i]

    def ioctl(self, op, arg):

            

Reported by Pylint.