# Copyright 2014 ETH Zurich
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
:mod:`lib_packet_scion_test` --- lib.packet.scion unit tests
============================================================
"""
# Stdlib
from unittest.mock import patch, MagicMock, call

# External packages
import capnp  # noqa
import nose
import nose.tools as ntools

# SCION
from lib.errors import SCIONIndexError, SCIONParseError
from lib.packet.ext_hdr import ExtensionHeader
from lib.packet.host_addr import HostAddrInvalidType
from lib.packet.packet_base import L4HeaderBase, PayloadRaw
from lib.packet.path import SCIONPath
from lib.packet.scion import (
    SCIONAddrHdr,
    SCIONBasePacket,
    SCIONCommonHdr,
    SCIONExtPacket,
    SCIONL4Packet,
    build_base_hdrs,
)
from lib.packet.scmp.errors import (
    SCMPBadDstType,
    SCMPBadEnd2End,
    SCMPBadHOFOffset,
    SCMPBadHopByHop,
    SCMPBadIOFOffset,
    SCMPBadPktLen,
    SCMPBadSrcType,
    SCMPBadVersion,
)
from lib.types import (
    AddrType,
    ExtHopByHopType,
    ExtensionClass,
    L4Proto,
    PayloadClass,
)
from test.testcommon import assert_these_calls, create_mock, create_mock_full


class TestSCIONCommonHdrParse(object):
    """
    Unit tests for lib.packet.scion.SCIONCommonHdr._parse
    """
    def _setup(self, first_b=0b00001111):
        inst = SCIONCommonHdr()
        data = create_mock(["pop"])
        data.pop.return_value = bytes([first_b, 0b00111111]) + \
            bytes.fromhex('0304 38 40 07 20')
        return inst, data

    @patch("lib.packet.scion.Raw", autospec=True)
    def test_bad_ver(self, raw):
        inst, data = self._setup(0b11110000)
        raw.return_value = data
        # Call
        ntools.assert_raises(SCMPBadVersion, inst._parse, "data")

    @patch("lib.packet.scion.SCIONAddrHdr.calc_lens", new_callable=create_mock)
    @patch("lib.packet.scion.Raw", autospec=True)
    def test_success(self, raw, calc_lens):
        inst, data = self._setup()
        raw.return_value = data
        calc_lens.return_value = 24, 0
        # Call
        inst._parse("data")
        # Tests
        raw.assert_called_once_with("data", inst.NAME, inst.LEN)
        ntools.eq_(inst.total_len, 0x0304)
        ntools.eq_(inst.next_hdr, 0x07)
        ntools.eq_(inst.hdr_len, 0x20)
        ntools.eq_(inst.version, 0b0)
        ntools.eq_(inst.src_addr_type, 0b111100)
        ntools.eq_(inst.dst_addr_type, 0b111111)
        calc_lens.assert_called_once_with(0b111100, 0b111111)
        ntools.eq_(inst.addrs_len, 24)
        ntools.eq_(inst._iof_idx, 3)
        ntools.eq_(inst._hof_idx, 4)

    @patch("lib.packet.scion.SCIONAddrHdr.calc_lens", new_callable=create_mock)
    @patch("lib.packet.scion.Raw", autospec=True)
    def test_bad_len(self, raw, calc_lens):
        inst, data = self._setup()
        calc_lens.return_value = 32, 0
        raw.return_value = data
        # Call
        ntools.assert_raises(SCIONParseError, inst._parse, "data")


class TestSCIONCommonHdrFromValues(object):
    """
    Unit tests for lib.packet.scion.SCIONCommonHdr.from_values
    """
    @patch("lib.packet.scion.SCIONAddrHdr.calc_lens", new_callable=create_mock)
    def test(self, calc_lens):
        # Setup
        src_type = 1
        dst_type = 2
        calc_lens.return_value = 44, 0
        hdr_len = SCIONCommonHdr.LEN + 44
        # Call
        inst = SCIONCommonHdr.from_values(src_type, dst_type, 3)
        # Tests
        ntools.assert_is_instance(inst, SCIONCommonHdr)
        ntools.eq_(inst.src_addr_type, src_type)
        ntools.eq_(inst.dst_addr_type, dst_type)
        calc_lens.assert_called_once_with(src_type, dst_type)
        ntools.eq_(inst.addrs_len, 44)
        ntools.eq_(inst.next_hdr, 3)
        ntools.eq_(inst.hdr_len, hdr_len)
        ntools.eq_(inst.total_len, hdr_len)


class TestSCIONCommonHdrPack(object):
    """
    Unit tests for lib.packet.scion.SCIONCommonHdr.pack
    """
    def test(self):
        inst = SCIONCommonHdr()
        inst.version = 0b1111
        inst.src_addr_type = 0b000000
        inst.dst_addr_type = 0b111111
        inst.addrs_len = 24
        inst.total_len = 0x304
        inst._iof_idx = 3
        inst._hof_idx = 4
        inst.next_hdr = 0x7
        inst.hdr_len = 0x8
        expected = b"".join([
            bytes([0b11110000, 0b00111111]),
            bytes.fromhex('0304 38 40 07 08'),
        ])
        # Call
        ntools.eq_(inst.pack(), expected)


class TestSCIONCommonHdrValidate(object):
    """
    Unit tests for lib.packet.scion.SCIONCommonHdr.validate
    """
    def _check(self, pkt_len, path_len, expected, iof=1, hof=3):
        inst = SCIONCommonHdr()
        inst.total_len = 10
        inst._iof_idx = iof
        inst._hof_idx = hof
        # Call
        if expected is None:
            inst.validate(pkt_len, path_len)
            return
        ntools.assert_raises(expected, inst.validate, pkt_len, path_len)

    def test_total_len(self):
        for pkt_len, expected in (
            (0, SCMPBadPktLen), (9, SCMPBadPktLen),
            (10, None), (11, SCMPBadPktLen),
        ):
            yield self._check, pkt_len, 10, expected

    def test_empty_path(self):
        for iof_idx, hof_idx, expected in (
            (0, 0, None), (1, 0, SCMPBadIOFOffset),
            (0, 1, SCMPBadHOFOffset), (1, 1, SCMPBadIOFOffset),
        ):
            yield self._check, 10, 0, expected, iof_idx, hof_idx

    def test_with_path(self):
        for hof_idx, expected in (
            (0, SCMPBadHOFOffset), (1, None)
        ):
            yield self._check, 10, 1, expected, 0, hof_idx


class TestSCIONCommonHdrStr(object):
    """
    Unit tests for lib.packet.scion.SCIONCommonHdr.__str__
    """
    @patch("lib.packet.scion.haddr_get_type", autospec=True)
    def test(self, get_type):
        inst = SCIONCommonHdr()
        inst.version = 0b1111
        inst.src_addr_type = 0b000000
        inst.dst_addr_type = 0b111111
        inst.addrs_len = 24
        inst.total_len = 0x304
        inst._iof_idx = 3
        inst._hof_idx = 4
        inst.next_hdr = 0x7
        inst.hdr_len = 0x8
        addr_type = create_mock(['name'])
        addr_type.name.return_value = "name"
        get_type.return_value = addr_type
        # Call
        str(inst)


class TestSCIONAddrHdrParse(object):
    """
    Unit tests for lib.packet.scion.SCIONAddrHdr._parse
    """
    def _setup(self, src_type):
        inst = SCIONAddrHdr()
        inst.calc_lens = create_mock()
        inst.update = create_mock()
        data = create_mock(["get", "pop"])
        data.get.side_effect = "src addr", "dst addr"
        data.pop.side_effect = None, None
        src = create_mock(["__len__", "host"])
        src.__len__.return_value = 3
        src.host = create_mock(["TYPE"])
        src.host.TYPE = src_type
        return inst, data, src, "dst"

    @patch("lib.packet.scion.SCIONAddr", autospec=True)
    @patch("lib.packet.scion.Raw", autospec=True)
    def test_success(self, raw, saddr):
        inst, data, src, dst = self._setup(AddrType.IPV4)
        raw.return_value = data
        saddr.side_effect = src, dst
        # Call
        inst._parse(1, 2, "data")
        # Tests
        inst.calc_lens.assert_called_once_with(1, 2)
        raw.assert_called_once_with(
            "data", inst.NAME, inst.calc_lens.return_value[0])
        assert_these_calls(
            saddr, [call((1, "src addr")), call((2, "dst addr"))])
        assert_these_calls(data.pop, [call(len(src)), call(len(dst))])
        ntools.eq_(inst.src, src)
        ntools.eq_(inst.dst, dst)
        inst.update.assert_called_once_with()

    @patch("lib.packet.scion.SCIONAddr", autospec=True)
    @patch("lib.packet.scion.Raw", autospec=True)
    def test_fail(self, raw, saddr):
        inst, data, src, dst = self._setup(AddrType.SVC)
        raw.return_value = data
        saddr.side_effect = src, dst
        # Call
        ntools.assert_raises(SCMPBadSrcType, inst._parse, 1, 2, "data")


class TestSCIONAddrHdrPack(object):
    """
    Unit tests for lib.packet.scion.SCIONAddrHdr.pack
    """
    def test(self):
        inst = SCIONAddrHdr()
        inst.update = create_mock()
        inst.src = create_mock(["pack"])
        inst.src.pack.return_value = b"src saddr"
        inst.dst = create_mock(["pack"])
        inst.dst.pack.return_value = b"dst saddr"
        inst._total_len = 24
        inst._pad_len = 6
        expected = b"src saddr" b"dst saddr"
        expected += bytes(inst._pad_len)
        # Call
        ntools.eq_(inst.pack(), expected)
        # Tests
        inst.update.assert_called_once_with()


class TestSCIONAddrHdrUpdate(object):
    """
    Unit tests for lib.packet.scion.SCIONAddrHdr.update
    """
    def test(self):
        inst = SCIONAddrHdr()
        inst.calc_lens = create_mock()
        inst.calc_lens.return_value = 1, 3
        inst.src = create_mock(["host"])
        inst.src.host = create_mock(["TYPE"])
        inst.dst = create_mock(["host"])
        inst.dst.host = create_mock(["TYPE"])
        # Call
        inst.update()
        # Tests
        inst.calc_lens.assert_called_once_with(
            inst.src.host.TYPE, inst.dst.host.TYPE)
        ntools.eq_(inst._total_len, 1)
        ntools.eq_(inst._pad_len, 3)


class TestSCIONAddrHdrCalcLens(object):
    """
    Unit tests for lib.packet.scion.SCIONAddrHdr.calc_lens
    """
    @patch("lib.packet.scion.calc_padding", autospec=True)
    @patch("lib.packet.scion.SCIONAddr.calc_len", new_callable=create_mock)
    def _check(self, type_lens, exp_total, exp_pad, calc_len, calc_padding):
        calc_len.side_effect = type_lens
        calc_padding.return_value = exp_pad
        # Call
        results = SCIONAddrHdr.calc_lens(1, 2)
        # Tests
        assert_these_calls(calc_len, [call(1), call(2)])
        calc_padding.assert_called_once_with(
            sum(type_lens), SCIONAddrHdr.BLK_SIZE)
        ntools.eq_(results, (exp_total, exp_pad))

    def test(self):
        for type_lens, exp_total, exp_pad in (
            ((8, 16), 24, 0),  # no padding required
            ((8, 19), 32, 5),  # padding required
        ):
            yield self._check, type_lens, exp_total, exp_pad

    @patch("lib.packet.scion.SCIONAddr.calc_len", new_callable=create_mock)
    def test_bad_src_type(self, calc_len):
        calc_len.side_effect = HostAddrInvalidType
        # Call
        ntools.assert_raises(SCMPBadSrcType, SCIONAddrHdr.calc_lens, 1, 2)

    @patch("lib.packet.scion.SCIONAddr.calc_len", new_callable=create_mock)
    def test_bad_dst_type(self, calc_len):
        calc_len.side_effect = "data len", HostAddrInvalidType
        # Call
        ntools.assert_raises(SCMPBadDstType, SCIONAddrHdr.calc_lens, 1, 2)


class TestSCIONAddrHdrReverse(object):
    """
    Unit tests for lib.packet.scion.SCIONAddrHdr.reverse
    """
    def test(self):
        inst = SCIONAddrHdr()
        inst.update = create_mock()
        inst.src = "src"
        inst.dst = "dst"
        # Call
        inst.reverse()
        # Tests
        ntools.eq_(inst.src, "dst")
        ntools.eq_(inst.dst, "src")
        inst.update.assert_called_once_with()


class TestSCIONAddrHdrLen(object):
    """
    Unit tests for lib.packet.scion.SCIONAddrHdr.__len__
    """
    def test(self):
        inst = SCIONAddrHdr()
        inst._total_len = 42
        # Call
        ntools.eq_(len(inst), 42)


class TestSCIONBasePacketParse(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket._parse
    """
    @patch("lib.packet.scion.PayloadRaw", autospec=True)
    @patch("lib.packet.scion.Raw", autospec=True)
    def test(self, raw, pld_raw):
        inst = SCIONBasePacket()
        inst._inner_parse = create_mock()
        inst.set_payload = create_mock()
        data = create_mock(["get"])
        raw.return_value = data
        # Call
        inst._parse(b"data")
        # Tests
        raw.assert_called_once_with(b"data", inst.NAME, inst.MIN_LEN, min_=True)
        inst._inner_parse.assert_called_once_with(data)
        pld_raw.assert_called_once_with(data.get.return_value)
        inst.set_payload.assert_called_once_with(pld_raw.return_value)


class TestSCIONBasePacketParseAddrs(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket._parse_addrs
    """
    @patch("lib.packet.scion.SCIONAddrHdr", autospec=True)
    def test(self, addr_hdr):
        inst = SCIONBasePacket()
        cmn_hdr = create_mock(["addrs_len", "src_addr_type", "dst_addr_type"])
        inst.cmn_hdr = cmn_hdr
        data = create_mock(["get", "pop"])
        # Call
        inst._parse_addrs(data)
        # Tests
        addr_hdr.assert_called_once_with((
            cmn_hdr.src_addr_type, cmn_hdr.dst_addr_type,
            data.get.return_value))
        ntools.eq_(inst.addrs, addr_hdr.return_value)
        data.pop.assert_called_once_with(addr_hdr.__len__.return_value)


class TestSCIONBasePacketParsePath(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket._parse_path
    """
    def _setup(self, data_offset=10, hdr_len=20):
        inst = SCIONBasePacket()
        inst.cmn_hdr = create_mock(["get_of_idxs", "hdr_len"])
        inst.cmn_hdr.get_of_idxs.return_value = "iof", "hof"
        inst.cmn_hdr.hdr_len = hdr_len
        data = create_mock(["__len__", "get", "offset", "pop"])
        data.__len__.return_value = 40 - data_offset
        data.offset.return_value = data_offset
        return inst, data

    @patch("lib.packet.scion.parse_path", autospec=True)
    def test_too_short(self, parse_path):
        inst, data = self._setup(hdr_len=9)
        # Call
        ntools.assert_raises(SCIONParseError, inst._parse_path, data)

    @patch("lib.packet.scion.parse_path", autospec=True)
    def test_too_long(self, parse_path):
        inst, data = self._setup(hdr_len=50)
        # Call
        ntools.assert_raises(SCIONParseError, inst._parse_path, data)

    @patch("lib.packet.scion.parse_path", autospec=True)
    def test_success(self, parse_path):
        inst, data = self._setup()
        path = create_mock(["__len__", "set_of_idxs"])
        path.__len__.return_value = 42
        parse_path.return_value = path
        # Call
        inst._parse_path(data)
        # Tests
        parse_path.assert_called_once_with(data.get.return_value)
        ntools.eq_(inst.path, path)
        data.pop.assert_called_once_with(42)
        path.set_of_idxs.assert_called_once_with("iof", "hof")


class TestSCIONBasePacketFromValues(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket.from_values
    """
    @patch("lib.packet.scion.SCIONBasePacket.update", autospec=True)
    @patch("lib.packet.scion.SCIONBasePacket.set_payload", autospec=True)
    @patch("lib.packet.scion.PayloadRaw", autospec=True)
    @patch("lib.packet.scion.SCIONBasePacket._inner_from_values", autospec=True)
    def test(self, inner_values, pld_raw, set_pld, update):
        inst = SCIONBasePacket.from_values("cmn hdr", "addr hdr", "path hdr")
        # Tests
        ntools.assert_is_instance(inst, SCIONBasePacket)
        inner_values.assert_called_once_with(
            inst, "cmn hdr", "addr hdr", "path hdr")
        pld_raw.assert_called_once_with()
        set_pld.assert_called_once_with(inst, pld_raw.return_value)
        update.assert_called_once_with(inst)


class TestSCIONBasePacketInnerFromValues(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket._inner_from_values
    """
    def test(self):
        inst = SCIONBasePacket()
        cmn_hdr = MagicMock(spec_set=SCIONCommonHdr)
        addr_hdr = MagicMock(spec_set=SCIONAddrHdr)
        path_hdr = MagicMock(spec_set=SCIONPath)
        # Call
        inst._inner_from_values(cmn_hdr, addr_hdr, path_hdr)


class TestSCIONBasePacketPack(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket.pack
    """
    @patch("lib.packet.scion.SCIONBasePacket.update", autospec=True)
    def test(self, update):
        hdr_exp = b"cmn hdr" b"addrs" b"path"
        total_exp = hdr_exp + b"inner pack"
        inst = SCIONBasePacket()
        inst.update = create_mock()
        inst.cmn_hdr = create_mock(["hdr_len", "pack", "total_len"])
        inst.cmn_hdr.hdr_len = len(hdr_exp)
        inst.cmn_hdr.total_len = len(total_exp)
        inst.cmn_hdr.pack.return_value = b"cmn hdr"
        inst.addrs = create_mock(["pack"])
        inst.addrs.pack.return_value = b"addrs"
        inst.path = create_mock(["pack"])
        inst.path.pack.return_value = b"path"
        inst._inner_pack = create_mock()
        inst._inner_pack.return_value = b"inner pack"
        # Call
        ntools.eq_(inst.pack(), total_exp)
        # Tests
        inst.update.assert_called_once_with()


class TestSCIONBasePacketValidate(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket.validate
    """
    def test(self):
        inst = SCIONBasePacket()
        inst.path = "path"
        inst.cmn_hdr = create_mock(["validate"])
        inst.addrs = create_mock(["validate"])
        inst._validate_of_idxes = create_mock()
        inst._payload = PayloadRaw()
        # Call
        inst.validate("pkt len")
        # Tests
        inst.cmn_hdr.validate.assert_called_once_with("pkt len", 4)
        inst.addrs.validate.assert_called_once_with()
        inst._validate_of_idxes.assert_called_once_with()


class TestSCIONBasePacketValidateOFIdxes(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket._validate_of_idxes
    """
    def test_bad_iof(self):
        inst = SCIONBasePacket()
        inst.path = create_mock(["get_iof"])
        inst.path.get_iof.side_effect = SCIONIndexError
        # Call
        ntools.assert_raises(SCMPBadIOFOffset, inst._validate_of_idxes)

    def test_bad_hof(self):
        inst = SCIONBasePacket()
        inst.path = create_mock(["get_iof", "get_hof"])
        inst.path.get_hof.side_effect = SCIONIndexError
        # Call
        ntools.assert_raises(SCMPBadHOFOffset, inst._validate_of_idxes)


class TestSCIONBasePacketUpdate(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket.update
    """
    def test(self):
        inst = SCIONBasePacket()
        inst.addrs = create_mock(["update"])
        inst._update_cmn_hdr = create_mock()
        # Call
        inst.update()
        # Tests
        inst.addrs.update.assert_called_once_with()
        inst._update_cmn_hdr.assert_called_once_with()


class TestSCIONBasePacketUpdateCmnHdr(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket._update_cmn_hdr
    """
    def test(self):
        inst = SCIONBasePacket()
        cmn_hdr = create_mock([
            "__len__", "src_addr_type", "dst_addr_type", "addrs_len", "hdr_len",
            "total_len", "set_of_idxs", "next_hdr", "update",
        ])
        cmn_hdr.__len__.return_value = 2
        addrs = create_mock(["__len__", "src_type", "dst_type"])
        addrs.__len__.return_value = 4
        path = create_mock(["__len__", "get_of_idxs"])
        path.__len__.return_value = 8
        path.get_of_idxs.return_value = 3, 7
        inst.cmn_hdr = cmn_hdr
        inst.addrs = addrs
        inst.path = path
        inst._get_offset_len = create_mock()
        inst._get_offset_len.return_value = 42
        inst._get_next_hdr = create_mock()
        # Call
        inst._update_cmn_hdr()
        # Tests
        ntools.eq_(cmn_hdr.src_addr_type, addrs.src_type.return_value)
        ntools.eq_(cmn_hdr.dst_addr_type, addrs.dst_type.return_value)
        ntools.eq_(cmn_hdr.addrs_len, 4)
        ntools.eq_(cmn_hdr.hdr_len, 2 + 4 + 8)
        ntools.eq_(cmn_hdr.total_len, 14 + 42)
        cmn_hdr.set_of_idxs.assert_called_once_with(3, 7)
        ntools.eq_(cmn_hdr.next_hdr, inst._get_next_hdr.return_value)


class TestSCIONBasePacketReverse(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket.reverse
    """
    def test(self):
        inst = SCIONBasePacket()
        inst.addrs = create_mock(['reverse'])
        inst.path = create_mock(['reverse'])
        # Call
        inst.reverse()
        # Tests
        inst.addrs.reverse.assert_called_once_with()
        inst.path.reverse.assert_called_once_with()


class TestSCIONBasePacketConvertToSCMPError(object):
    """
    Unit tests for lib.packet.scion.SCIONBasePacket.convert_to_scmp_error
    """
    @patch("lib.packet.scion.SCMPHeader.from_values", new_callable=create_mock)
    @patch("lib.packet.scion.SCMPPayload.from_pkt", new_callable=create_mock)
    @patch("lib.packet.scion.SCMPExt.from_values", new_callable=create_mock)
    def test(self, ext_fv, pld_fp, hdr_fv):
        inst = SCIONBasePacket()
        inst.addrs = create_mock(["dst", "src"])
        scmp_ext = create_mock(["EXT_TYPE"])
        scmp_ext.EXT_TYPE = ExtHopByHopType.SCMP
        inst.ext_hdrs = [scmp_ext, "ext1", "ext2", "ext3", "ext4"]
        inst.set_payload = create_mock()
        # Call
        inst.convert_to_scmp_error(
            "addr", "class", "type", "pkt", "arg1", "arg2", hopbyhop="hbh",
            kwarg1="kwval1")
        # Tests
        ext_fv.assert_called_once_with(hopbyhop="hbh")
        ntools.eq_(inst.ext_hdrs, [ext_fv.return_value, "ext1", "ext2", "ext3"])
        pld_fp.assert_called_once_with(
            "class", "type", "pkt", "arg1", "arg2", kwarg1="kwval1")
        hdr_fv.assert_called_once_with(
            "addr", inst.addrs.dst, "class", "type")
        ntools.eq_(inst.l4_hdr, hdr_fv.return_value)
        inst.set_payload.assert_called_once_with(pld_fp.return_value)


class TestSCIONExtPacketFromValues(object):
    """
    Unit tests for lib.packet.scion.SCIONExtPacket.from_values
    """
    @patch("lib.packet.scion.SCIONExtPacket.set_payload", autospec=True)
    @patch("lib.packet.scion.SCIONExtPacket._inner_from_values", autospec=True)
    def test_basic(self, inner_values, set_pld):
        inst = SCIONExtPacket.from_values("cmn hdr", "addr hdr", "path hdr",
                                          "ext hdrs")
        # Tests
        ntools.assert_is_instance(inst, SCIONExtPacket)
        inner_values.assert_called_once_with(
            inst, "cmn hdr", "addr hdr", "path hdr", "ext hdrs")
        set_pld.assert_called_once_with(inst, b"")

    @patch("lib.packet.scion.SCIONExtPacket.set_payload", autospec=True)
    @patch("lib.packet.scion.SCIONExtPacket._inner_from_values", autospec=True)
    def test_payload(self, inner_values, set_pld):
        inst = SCIONExtPacket.from_values("cmn hdr", "addr hdr", "path hdr",
                                          "ext hdrs", "payload")
        # Tests
        set_pld.assert_called_once_with(inst, "payload")


class TestSCIONExtPacketInnerFromValues(object):
    """
    Unit tests for lib.packet.scion.SCIONExtPacket._inner_from_values
    """
    @patch("lib.packet.scion.SCIONBasePacket._inner_from_values", autospec=True)
    def test(self, super_values):
        inst = SCIONExtPacket()
        ext_hdrs = []
        for _ in range(3):
            ext_hdrs.append(MagicMock(spec_set=ExtensionHeader))
        # Call
        inst._inner_from_values("cmn hdr", "addr hdr", "path hdr", ext_hdrs)
        # Tests
        super_values.assert_called_once_with(
            inst, "cmn hdr", "addr hdr", "path hdr")
        ntools.eq_(inst.ext_hdrs, ext_hdrs)


class TestSCIONExtPacketInnerPack(object):
    """
    Unit tests for lib.packet.scion.SCIONExtPacket._inner_pack
    """
    @patch("lib.packet.scion.SCIONBasePacket._inner_pack", autospec=True)
    def test(self, super_pack):
        inst = SCIONExtPacket()
        inst._l4_proto = 0x42
        super_pack.return_value = b"super"
        inst.ext_hdrs = []
        for idx, class_, len_, type_ in (
            (0, 0x1, 0x0, 0x0), (1, 0x2, 0x1, 0x11), (2, 0x3, 0x2, 0x1)
        ):
            hdr = create_mock(["hdr_len", "EXT_CLASS", "EXT_TYPE", "pack"])
            hdr.EXT_CLASS = class_
            hdr.hdr_len.return_value = len_
            hdr.EXT_TYPE = type_
            hdr.pack.return_value = bytes(
                range(5+len_*ExtensionHeader.LINE_LEN))
            inst.ext_hdrs.append(hdr)
        expected = b"".join([
            b"super",
            bytes([0x2, 0x0, 0x0]), bytes(range(5)),
            bytes([0x3, 0x1, 0x11]), bytes(range(13)),
            bytes([0x42, 0x2, 0x1]), bytes(range(21)),
        ])
        # Call
        ntools.eq_(inst._inner_pack(), expected)


class TestSCIONExtPacketGetOffsetLen(object):
    """
    Unit tests for lib.packet.scion.SCIONExtPacket._get_offset_len
    """
    @patch("lib.packet.scion.SCIONBasePacket._get_offset_len", autospec=True)
    def test(self, super_offset):
        inst = SCIONExtPacket()
        inst.ext_hdrs = ["a", "bb", "cccc"]
        super_offset.return_value = 42
        # Call
        ntools.eq_(inst._get_offset_len(), 49)


class TestSCIONExtPacketGetNextHdr(object):
    """
    Unit tests for lib.packet.scion.SCIONExtPacket._get_next_hdr
    """
    def test_no_exts(self):
        inst = SCIONExtPacket()
        inst.ext_hdrs = []
        inst._l4_proto = 42
        # Call
        ntools.eq_(inst._get_next_hdr(), 42)

    def test_with_exts(self):
        inst = SCIONExtPacket()
        inst.ext_hdrs = [create_mock(["EXT_CLASS"]), 0]
        # Call
        ntools.eq_(inst._get_next_hdr(), inst.ext_hdrs[0].EXT_CLASS)


class TestSCIONExtPacketValidate(object):
    """
    Unit tests for lib.packet.scion.SCIONExtPacket.validate
    """
    @patch("lib.packet.scion.SCIONBasePacket.validate", autospec=True)
    def test_no_unknown(self, super_val):
        inst = SCIONExtPacket()
        # Call
        ntools.ok_(inst.validate("pkt len"))
        # Tests
        super_val.assert_called_once_with(inst, "pkt len")

    @patch("lib.packet.scion.SCIONBasePacket.validate", autospec=True)
    def test_unknown_hbh(self, super_val):
        inst = SCIONExtPacket()
        inst._unknown_exts = {ExtensionClass.HOP_BY_HOP: [1]}
        # Call
        ntools.assert_raises(SCMPBadHopByHop, inst.validate, "pkt len")

    @patch("lib.packet.scion.SCIONBasePacket.validate", autospec=True)
    def test_unknown_e2e(self, super_val):
        inst = SCIONExtPacket()
        inst._unknown_exts = {ExtensionClass.END_TO_END: [1]}
        # Call
        ntools.assert_raises(SCMPBadEnd2End, inst.validate, "pkt len")


class TestSCIONL4PacketInnerParse(object):
    """
    Unit tests for lib.packet.scion.SCIONL4Packet._inner_parse
    """
    @patch("lib.packet.scion.parse_l4_hdr", autospec=True)
    @patch("lib.packet.scion.SCIONExtPacket._inner_parse", autospec=True)
    def test(self, super_parse, parse_l4_hdr):
        inst = SCIONL4Packet()
        inst._l4_proto = 42
        inst.addrs = create_mock(["src", "dst"])
        data = create_mock()
        parse_l4_hdr.return_value.__len__.return_value = 8
        # Call
        inst._inner_parse(data)
        # Tests
        super_parse.assert_called_once_with(inst, data)
        parse_l4_hdr.assert_called_once_with(
            42, data, src=inst.addrs.src, dst=inst.addrs.dst,
        )
        ntools.eq_(inst.l4_hdr, parse_l4_hdr.return_value)


class TestSCIONL4PacketFromValues(object):
    """
    Unit tests for lib.packet.scion.SCIONL4Packet.from_values
    """
    @patch("lib.packet.scion.SCIONL4Packet.update", autospec=True)
    @patch("lib.packet.scion.SCIONL4Packet.set_payload", autospec=True)
    @patch("lib.packet.scion.PayloadRaw", autospec=True)
    @patch("lib.packet.scion.SCIONL4Packet._inner_from_values", autospec=True)
    def test(self, inner_values, pldraw, set_pld, upd_hdrs):
        inst = SCIONL4Packet.from_values("cmn hdr", "addr hdr", "path hdr",
                                         "ext hdrs", "l4 hdr")
        # Tests
        ntools.assert_is_instance(inst, SCIONL4Packet)
        inner_values.assert_called_once_with(
            inst, "cmn hdr", "addr hdr", "path hdr", "ext hdrs", "l4 hdr")
        set_pld.assert_called_once_with(inst, pldraw.return_value)
        upd_hdrs.assert_called_once_with(inst)


class TestSCIONL4PacketInnerFromValues(object):
    """
    Unit tests for lib.packet.scion.SCIONL4Packet._inner_from_values
    """
    @patch("lib.packet.scion.SCIONExtPacket._inner_from_values", autospec=True)
    def test(self, super_values):
        inst = SCIONL4Packet()
        l4_hdr = MagicMock(spec_set=L4HeaderBase)
        l4_hdr.TYPE = 22
        # Call
        inst._inner_from_values("cmn hdr", "addr hdr", "path hdr", "ext hdrs",
                                l4_hdr)
        # Tests
        super_values.assert_called_once_with(
            inst, "cmn hdr", "addr hdr", "path hdr", "ext hdrs")
        ntools.eq_(inst.l4_hdr, l4_hdr)
        ntools.eq_(inst._l4_proto, 22)


class TestSCIONL4PacketInnerPack(object):
    """
    Unit tests for lib.packet.scion.SCIONL4Packet._inner_pack
    """
    @patch("lib.packet.scion.SCIONExtPacket._pack_payload", autospec=True)
    @patch("lib.packet.scion.SCIONExtPacket._inner_pack", autospec=True)
    def test(self, super_pack, super_pld):
        inst = SCIONL4Packet()
        inst.update = create_mock()
        super_pack.return_value = b"super"
        super_pld.return_value = b"pld"
        inst.l4_hdr = create_mock(["pack"])
        inst.l4_hdr.pack.return_value = b"l4 hdr"
        # Call
        ntools.eq_(inst._inner_pack(), b"super" b"l4 hdr" b"pld")
        # Tests
        inst.update.assert_called_once_with()
        inst.l4_hdr.pack.assert_called_once_with(b"pld")


class TestSCIONL4PacketUpdate(object):
    """
    Unit tests for lib.packet.scion.SCIONL4Packet.update
    """
    @patch("lib.packet.scion.SCIONExtPacket.update", autospec=True)
    def test_no_l4hdr(self, super_update):
        inst = SCIONL4Packet()
        inst._l4_proto = 47
        # Call
        inst.update()
        # Tests
        super_update.assert_called_once_with(inst)
        ntools.eq_(inst._l4_proto, 47)

    @patch("lib.packet.scion.SCIONExtPacket.update", autospec=True)
    def test_l4hdr(self, super_update):
        inst = SCIONL4Packet()
        inst.addrs = create_mock(["src", "dst"])
        inst._l4_proto = 47
        inst.l4_hdr = create_mock(["TYPE", "update"])
        # Call
        inst.update()
        # Tests
        inst.l4_hdr.update.assert_called_once_with(
            src=inst.addrs.src, dst=inst.addrs.dst)
        ntools.eq_(inst._l4_proto, inst.l4_hdr.TYPE)


class TestSCIONL4PacketParsePayload(object):
    """
    Unit tests for lib.packet.scion.SCIONL4Packet.parse_payload
    """
    def _setup(self, l4_type):
        inst = SCIONL4Packet()
        inst._parse_pld_ctrl = create_mock()
        inst._parse_pld_ctrl.return_value = "pld ctrl"
        inst._parse_pld_scmp = create_mock()
        inst._parse_pld_scmp.return_value = "pld scmp"
        inst.set_payload = create_mock()
        inst._payload = create_mock(["pack"])
        if l4_type is not None:
            inst.l4_hdr = create_mock(["TYPE"])
            inst.l4_hdr.TYPE = l4_type
        return inst

    @patch("lib.packet.scion.Raw", autospec=True)
    def test_non_l4(self, raw):
        inst = self._setup(None)
        # Call
        ntools.assert_raises(SCIONParseError, inst.parse_payload)
        # Tests
        raw.assert_called_once_with(inst._payload.pack.return_value,
                                    "SCIONL4Packet.parse_payload")

    @patch("lib.packet.scion.Raw", autospec=True)
    def _check_l4(self, l4_type, expected, raw):
        inst = self._setup(l4_type)
        # Call
        ntools.eq_(inst.parse_payload(), expected)
        # Tests
        inst.set_payload.assert_called_once_with(expected)
        if l4_type == L4Proto.UDP:
            inst._parse_pld_ctrl.assert_called_once_with(raw.return_value)
            ntools.assert_false(inst._parse_pld_scmp.called)
        elif l4_type == L4Proto.SCMP:
            inst._parse_pld_scmp.assert_called_once_with(raw.return_value)
            ntools.assert_false(inst._parse_pld_ctrl.called)

    def test_l4(self):
        for l4_type, exp in (
            (L4Proto.UDP, "pld ctrl"),
            (L4Proto.SCMP, "pld scmp"),
        ):
            yield self._check_l4, l4_type, exp


class TestSCIONL4PacketParsePldCtrl(object):
    """
    Unit tests for lib.packet.scion.SCIONL4Packet._parse_pld_ctrl
    """
    def _mk_data(self, len_=4):
        return create_mock_full({
            "pop()...": [bytes((0, 0, 0, 4)), "outer"], "__len__()": len_
        })

    def test_bad_len(self):
        inst = SCIONL4Packet()
        data = self._mk_data(3)
        # Call
        ntools.assert_raises(SCIONParseError, inst._parse_pld_ctrl, data)

    def _raise_parse_error(self, _):
        raise capnp.lib.capnp.KjException("error")

    @patch("lib.packet.scion.P.SCION.from_bytes_packed", autospec=True)
    def test_parse_error(self, from_bytes):
        inst = SCIONL4Packet()
        data = self._mk_data(4)
        from_bytes.side_effect = self._raise_parse_error
        # Call
        ntools.assert_raises(SCIONParseError, inst._parse_pld_ctrl, data)

    def _mk_proto_obj(self, class_):
        builder = create_mock_full({"which()": class_, class_: "inner"})
        return create_mock_full({"as_builder()": builder})

    @patch("lib.packet.scion.parse_sibra_payload", autospec=True)
    @patch("lib.packet.scion.parse_pathmgmt_payload", autospec=True)
    @patch("lib.packet.scion.parse_certmgmt_payload", autospec=True)
    @patch("lib.packet.scion.parse_ifid_payload", autospec=True)
    @patch("lib.packet.scion.parse_pcb_payload", autospec=True)
    @patch("lib.packet.scion.P.SCION.from_bytes_packed", autospec=True)
    def _check_known(self, class_, from_bytes, parse_pcb, parse_ifid,
                     parse_cert, parse_path, parse_sibra):
        from_bytes.return_value = self._mk_proto_obj(class_)
        class_map = {
            PayloadClass.PCB: parse_pcb, PayloadClass.IFID: parse_ifid,
            PayloadClass.CERT: parse_cert, PayloadClass.PATH: parse_path,
            PayloadClass.SIBRA: parse_sibra,
        }
        handler = class_map[class_]
        inst = SCIONL4Packet()
        data = self._mk_data()
        # Call
        ntools.eq_(inst._parse_pld_ctrl(data), handler.return_value)
        # Tests
        handler.assert_called_once_with("inner")

    def test_known(self):
        for class_ in (
            PayloadClass.PCB, PayloadClass.IFID, PayloadClass.CERT,
            PayloadClass.PATH, PayloadClass.SIBRA,
        ):
            yield self._check_known, class_

    @patch("lib.packet.scion.P.SCION.from_bytes_packed", autospec=True)
    def test_unknown(self, from_bytes):
        from_bytes.return_value = self._mk_proto_obj("unknown")
        inst = SCIONL4Packet()
        data = self._mk_data()
        # Call
        ntools.assert_raises(SCIONParseError, inst._parse_pld_ctrl, data)


class TestSCIONL4PacketGetOffsetLen(object):
    """
    Unit tests for lib.packet.scion.SCIONL4Packet._get_offset_len
    """
    @patch("lib.packet.scion.SCIONExtPacket._get_offset_len", autospec=True)
    def test(self, super_offset):
        inst = SCIONL4Packet()
        inst.l4_hdr = create_mock(["total_len"])
        inst.l4_hdr.total_len = 12
        super_offset.return_value = 42
        # Call
        ntools.eq_(inst._get_offset_len(), 54)


class TestBuildBaseHdrs(object):
    """
    Unit tests for lib.packet.scion.build_base_hdrs
    """
    @patch("lib.packet.scion.SCIONAddrHdr.from_values",
           new_callable=create_mock)
    @patch("lib.packet.scion.SCIONCommonHdr.from_values",
           new_callable=create_mock)
    def test(self, cmn_hdr, addr_hdr):
        src = create_mock(["host"])
        src.host = create_mock(["TYPE"])
        dst = create_mock(["host"])
        dst.host = create_mock(["TYPE"])
        # Call
        ntools.eq_(build_base_hdrs(src, dst),
                   (cmn_hdr.return_value, addr_hdr.return_value))
        # Tests
        cmn_hdr.assert_called_once_with(src.host.TYPE, dst.host.TYPE,
                                        L4Proto.UDP)
        addr_hdr.assert_called_once_with(src, dst)


if __name__ == "__main__":
    nose.run(defaultTest=__name__)
