# Copyright (C) 2018-2018, Semtech Corporation.
# All rights reserved.
#
# This file is subject to the terms and conditions
# defined in file 'LICENSE', which is part of this
# source code package.

from typing import Optional,Tuple

import argparse
import asyncio
import struct
import sys

from devsimul import LoraMsg, Simulation
from devtest import DeviceTest, LoRaWANTest
from colorama import init as colorama_init
from vtimeloop import VirtualTimeLoop

class FWMan:
    PORT        = 203
    PKG_ID      = 4
    PKG_VERSION = 0x00
    DEV_VERSION = 0x01

class Frag:
    PORT            = 201
    PKG_ID          = 3
    PKG_VERSION     = 0x00
    FRAG_STATUS     = 0x01
    FRAG_SESS_SETUP = 0x02
    FRAG_SESS_DEL   = 0x03
    DATA_FRAGMENT   = 0x08


class FuotaTest(LoRaWANTest):
    @DeviceTest.test(reboot=False)
    async def join(self) -> bool:
        msg = await self.upmsg()
        self.process_join(msg)

        # expect any message
        msg = await self.upmsg()
        m = self.verify(msg)

        return True

    @DeviceTest.test(reboot=False)
    async def fwman_pkgversion(self) -> bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.PKG_VERSION))

        msg, m = await self.ul(FWMan.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == FWMan.PKG_VERSION and p == FWMan.PKG_ID and v == 1

        return True

    @DeviceTest.test(reboot=False)
    async def fwman_devversion(self) -> bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.DEV_VERSION))

        msg, m = await self.ul(FWMan.PORT)
        c, f, h = struct.unpack('<BII', m['FRMPayload'])
        assert c == FWMan.DEV_VERSION
        print('FW Version: 0x%08x, HW Version: 0x%08x' % (f, h))

        return True

    @DeviceTest.test(reboot=False)
    async def frag_pkgversion(self) -> bool:
        msg = await self.upmsg()
        self.dl(msg, Frag.PORT, struct.pack('B', Frag.PKG_VERSION))

        msg, m = await self.ul(Frag.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == Frag.PKG_VERSION and p == Frag.PKG_ID and v == 1

        return True


if __name__ == '__main__':
    p = argparse.ArgumentParser()
    LoRaWANTest.stdargs(p)
    args = p.parse_args()

    colorama_init()

    if args.virtual_time:
        asyncio.set_event_loop(VirtualTimeLoop()) # type: ignore

    sim = Simulation(args.hexfiles, args.debug, args.traffic)
    test = FuotaTest(sim, args)

    if not asyncio.get_event_loop().run_until_complete(test.run()):
        sys.exit(1)

4ee76b88b1d62669441def4e908b626ceafe65f1
<mask token>


class FWMan:
    <mask token>
    <mask token>
    <mask token>
    <mask token>


class Frag:
    PORT = 201
    PKG_ID = 3
    PKG_VERSION = 0
    FRAG_STATUS = 1
    FRAG_SESS_SETUP = 2
    FRAG_SESS_DEL = 3
    DATA_FRAGMENT = 8


class FuotaTest(LoRaWANTest):

    @DeviceTest.test(reboot=False)
    async def join(self) ->bool:
        msg = await self.upmsg()
        self.process_join(msg)
        msg = await self.upmsg()
        m = self.verify(msg)
        return True

    @DeviceTest.test(reboot=False)
    async def fwman_pkgversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.PKG_VERSION))
        msg, m = await self.ul(FWMan.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == FWMan.PKG_VERSION and p == FWMan.PKG_ID and v == 1
        return True

    @DeviceTest.test(reboot=False)
    async def fwman_devversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.DEV_VERSION))
        msg, m = await self.ul(FWMan.PORT)
        c, f, h = struct.unpack('<BII', m['FRMPayload'])
        assert c == FWMan.DEV_VERSION
        print('FW Version: 0x%08x, HW Version: 0x%08x' % (f, h))
        return True

    @DeviceTest.test(reboot=False)
    async def frag_pkgversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, Frag.PORT, struct.pack('B', Frag.PKG_VERSION))
        msg, m = await self.ul(Frag.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == Frag.PKG_VERSION and p == Frag.PKG_ID and v == 1
        return True


<mask token>

<mask token>


class FWMan:
    PORT = 203
    PKG_ID = 4
    PKG_VERSION = 0
    DEV_VERSION = 1


class Frag:
    PORT = 201
    PKG_ID = 3
    PKG_VERSION = 0
    FRAG_STATUS = 1
    FRAG_SESS_SETUP = 2
    FRAG_SESS_DEL = 3
    DATA_FRAGMENT = 8


class FuotaTest(LoRaWANTest):

    @DeviceTest.test(reboot=False)
    async def join(self) ->bool:
        msg = await self.upmsg()
        self.process_join(msg)
        msg = await self.upmsg()
        m = self.verify(msg)
        return True

    @DeviceTest.test(reboot=False)
    async def fwman_pkgversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.PKG_VERSION))
        msg, m = await self.ul(FWMan.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == FWMan.PKG_VERSION and p == FWMan.PKG_ID and v == 1
        return True

    @DeviceTest.test(reboot=False)
    async def fwman_devversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.DEV_VERSION))
        msg, m = await self.ul(FWMan.PORT)
        c, f, h = struct.unpack('<BII', m['FRMPayload'])
        assert c == FWMan.DEV_VERSION
        print('FW Version: 0x%08x, HW Version: 0x%08x' % (f, h))
        return True

    @DeviceTest.test(reboot=False)
    async def frag_pkgversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, Frag.PORT, struct.pack('B', Frag.PKG_VERSION))
        msg, m = await self.ul(Frag.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == Frag.PKG_VERSION and p == Frag.PKG_ID and v == 1
        return True


<mask token>

<mask token>


class FWMan:
    PORT = 203
    PKG_ID = 4
    PKG_VERSION = 0
    DEV_VERSION = 1


class Frag:
    PORT = 201
    PKG_ID = 3
    PKG_VERSION = 0
    FRAG_STATUS = 1
    FRAG_SESS_SETUP = 2
    FRAG_SESS_DEL = 3
    DATA_FRAGMENT = 8


class FuotaTest(LoRaWANTest):

    @DeviceTest.test(reboot=False)
    async def join(self) ->bool:
        msg = await self.upmsg()
        self.process_join(msg)
        msg = await self.upmsg()
        m = self.verify(msg)
        return True

    @DeviceTest.test(reboot=False)
    async def fwman_pkgversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.PKG_VERSION))
        msg, m = await self.ul(FWMan.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == FWMan.PKG_VERSION and p == FWMan.PKG_ID and v == 1
        return True

    @DeviceTest.test(reboot=False)
    async def fwman_devversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.DEV_VERSION))
        msg, m = await self.ul(FWMan.PORT)
        c, f, h = struct.unpack('<BII', m['FRMPayload'])
        assert c == FWMan.DEV_VERSION
        print('FW Version: 0x%08x, HW Version: 0x%08x' % (f, h))
        return True

    @DeviceTest.test(reboot=False)
    async def frag_pkgversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, Frag.PORT, struct.pack('B', Frag.PKG_VERSION))
        msg, m = await self.ul(Frag.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == Frag.PKG_VERSION and p == Frag.PKG_ID and v == 1
        return True


if __name__ == '__main__':
    p = argparse.ArgumentParser()
    LoRaWANTest.stdargs(p)
    args = p.parse_args()
    colorama_init()
    if args.virtual_time:
        asyncio.set_event_loop(VirtualTimeLoop())
    sim = Simulation(args.hexfiles, args.debug, args.traffic)
    test = FuotaTest(sim, args)
    if not asyncio.get_event_loop().run_until_complete(test.run()):
        sys.exit(1)

from typing import Optional, Tuple
import argparse
import asyncio
import struct
import sys
from devsimul import LoraMsg, Simulation
from devtest import DeviceTest, LoRaWANTest
from colorama import init as colorama_init
from vtimeloop import VirtualTimeLoop


class FWMan:
    PORT = 203
    PKG_ID = 4
    PKG_VERSION = 0
    DEV_VERSION = 1


class Frag:
    PORT = 201
    PKG_ID = 3
    PKG_VERSION = 0
    FRAG_STATUS = 1
    FRAG_SESS_SETUP = 2
    FRAG_SESS_DEL = 3
    DATA_FRAGMENT = 8


class FuotaTest(LoRaWANTest):

    @DeviceTest.test(reboot=False)
    async def join(self) ->bool:
        msg = await self.upmsg()
        self.process_join(msg)
        msg = await self.upmsg()
        m = self.verify(msg)
        return True

    @DeviceTest.test(reboot=False)
    async def fwman_pkgversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.PKG_VERSION))
        msg, m = await self.ul(FWMan.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == FWMan.PKG_VERSION and p == FWMan.PKG_ID and v == 1
        return True

    @DeviceTest.test(reboot=False)
    async def fwman_devversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.DEV_VERSION))
        msg, m = await self.ul(FWMan.PORT)
        c, f, h = struct.unpack('<BII', m['FRMPayload'])
        assert c == FWMan.DEV_VERSION
        print('FW Version: 0x%08x, HW Version: 0x%08x' % (f, h))
        return True

    @DeviceTest.test(reboot=False)
    async def frag_pkgversion(self) ->bool:
        msg = await self.upmsg()
        self.dl(msg, Frag.PORT, struct.pack('B', Frag.PKG_VERSION))
        msg, m = await self.ul(Frag.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == Frag.PKG_VERSION and p == Frag.PKG_ID and v == 1
        return True


if __name__ == '__main__':
    p = argparse.ArgumentParser()
    LoRaWANTest.stdargs(p)
    args = p.parse_args()
    colorama_init()
    if args.virtual_time:
        asyncio.set_event_loop(VirtualTimeLoop())
    sim = Simulation(args.hexfiles, args.debug, args.traffic)
    test = FuotaTest(sim, args)
    if not asyncio.get_event_loop().run_until_complete(test.run()):
        sys.exit(1)

# Copyright (C) 2018-2018, Semtech Corporation.
# All rights reserved.
#
# This file is subject to the terms and conditions
# defined in file 'LICENSE', which is part of this
# source code package.

from typing import Optional,Tuple

import argparse
import asyncio
import struct
import sys

from devsimul import LoraMsg, Simulation
from devtest import DeviceTest, LoRaWANTest
from colorama import init as colorama_init
from vtimeloop import VirtualTimeLoop

class FWMan:
    PORT        = 203
    PKG_ID      = 4
    PKG_VERSION = 0x00
    DEV_VERSION = 0x01

class Frag:
    PORT            = 201
    PKG_ID          = 3
    PKG_VERSION     = 0x00
    FRAG_STATUS     = 0x01
    FRAG_SESS_SETUP = 0x02
    FRAG_SESS_DEL   = 0x03
    DATA_FRAGMENT   = 0x08


class FuotaTest(LoRaWANTest):
    @DeviceTest.test(reboot=False)
    async def join(self) -> bool:
        msg = await self.upmsg()
        self.process_join(msg)

        # expect any message
        msg = await self.upmsg()
        m = self.verify(msg)

        return True

    @DeviceTest.test(reboot=False)
    async def fwman_pkgversion(self) -> bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.PKG_VERSION))

        msg, m = await self.ul(FWMan.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == FWMan.PKG_VERSION and p == FWMan.PKG_ID and v == 1

        return True

    @DeviceTest.test(reboot=False)
    async def fwman_devversion(self) -> bool:
        msg = await self.upmsg()
        self.dl(msg, FWMan.PORT, struct.pack('B', FWMan.DEV_VERSION))

        msg, m = await self.ul(FWMan.PORT)
        c, f, h = struct.unpack('<BII', m['FRMPayload'])
        assert c == FWMan.DEV_VERSION
        print('FW Version: 0x%08x, HW Version: 0x%08x' % (f, h))

        return True

    @DeviceTest.test(reboot=False)
    async def frag_pkgversion(self) -> bool:
        msg = await self.upmsg()
        self.dl(msg, Frag.PORT, struct.pack('B', Frag.PKG_VERSION))

        msg, m = await self.ul(Frag.PORT)
        c, p, v = struct.unpack('BBB', m['FRMPayload'])
        assert c == Frag.PKG_VERSION and p == Frag.PKG_ID and v == 1

        return True


if __name__ == '__main__':
    p = argparse.ArgumentParser()
    LoRaWANTest.stdargs(p)
    args = p.parse_args()

    colorama_init()

    if args.virtual_time:
        asyncio.set_event_loop(VirtualTimeLoop()) # type: ignore

    sim = Simulation(args.hexfiles, args.debug, args.traffic)
    test = FuotaTest(sim, args)

    if not asyncio.get_event_loop().run_until_complete(test.run()):
        sys.exit(1)
