regression.uts

% Regression tests for Scapy

# More informations at http://www.secdev.org/projects/UTscapy/

############
############
+ Informations on Scapy

= Get conf
~ conf command
* Dump the current configuration
conf

= List layers
~ conf command
ls()

= List commands
~ conf command
lsc()

= Configuration
~ conf
conf.debug_dissector = True


############
############
+ Scapy functions tests

= Interface related functions

get_if_raw_hwaddr(conf.iface)

get_if_raw_addr(conf.iface).encode("hex")

def get_dummy_interface():
    """Returns a dummy network interface"""
    if WINDOWS:
        data = {}
        data["name"] = "dummy0"
        data["description"] = "Does not exist"
        data["win_index"] = -1
        data["guid"] = "{0XX00000-X000-0X0X-X00X-00XXXX000XXX}"
        data["invalid"] = True
        return NetworkInterface(data)
    else:
        return "dummy0"

get_if_raw_addr(get_dummy_interface())

get_if_list()

get_if_raw_addr6(conf.iface6)

= Test read_routes6() - default output

routes6 = read_routes6()
if WINDOWS:
    route_add_loopback(routes6, True)

# Expected results:
# - one route if there is only the loopback interface
# - three routes if there is a network interface

if len(routes6):
    iflist = get_if_list()
    if WINDOWS:
        route_add_loopback(ipv6=True, iflist=iflist)
    if iflist == [LOOPBACK_NAME]:
	len(routes6) == 1
    elif len(iflist) >= 2:
	len(routes6) >= 3
    else:
	False
else:
    # IPv6 seems disabled. Force a route to ::1
    conf.route6.routes.append(("::1", 128, "::", LOOPBACK_NAME, ["::1"]))
    True

= Test read_routes6() - check mandatory routes

if len(routes6):
    assert(len(filter(lambda r: r[0] == "::1" and r[-1] == ["::1"], routes6)) >= 1)
    if iflist >= 2:
	assert(len(filter(lambda r: r[0] == "fe80::" and r[1] == 64, routes6)) >= 1)
	len(filter(lambda r: in6_islladdr(r[0]) and r[1] == 128 and r[-1] == ["::1"], routes6)) >= 1
else:
    True

= Test ifchange()
conf.route6.ifchange(LOOPBACK_NAME, "::1/128")
True

############
############
+ Main.py tests

= Prepare emulator
import mock
from mock import Mock

if WINDOWS:
    # This fix when the windows doesn't have a size (run with a windows server)
    mock.patch("pyreadline.console.console.Console.size").return_value = (300, 300)

@mock.patch("scapy.config.conf.readfunc")
def emulate_main_input(data, mock_readfunc):
    global index
    index = 0 # reset var
    def readlineScapy(*args, **kargs):
        global index
        if len(data) == index:
            r_data = "exit(1 if hasattr(sys, 'last_value') and sys.last_value is not None else 0)"
        else:
            r_data = data[index]
        index +=1
        print r_data
        return r_data
    mock_readfunc.side_effect = readlineScapy
    def reduce_mock(self):
        return (Mock, ())
    mock_readfunc.__reduce__ = reduce_mock
    sys.argv = ['']
    def console_exit(code):
        raise SystemExit(code)
    exit_code = -1
    try:
        interact(mydict={"exit": console_exit})
    except SystemExit as e:
        exit_code = str(e)
        pass
    assert exit_code == "0"

= Test basic save_session and load_session

data = ["init_session(\"scapySession1\")",\
"test_value = \"8.8.8.8\"",\
"save_session()",\
"del test_value",\
"load_session()",\
"assert test_value == \"8.8.8.8\""]

emulate_main_input(data)

= Test save_session and load_session with fname

data = ["init_session(\"scapySession2\")",\
"test_value = 7",\
"save_session(fname=\"scapySaveSession.dat\")",\
"del test_value",\
"load_session(fname=\"scapySaveSession.dat\")",\
"assert test_value == 7"]

emulate_main_input(data)

= Test utility functions

tmpfile =  get_temp_file(autoext=".ut")
tmpfile.startswith("/tmp/scapy")
conf.temp_files[0].endswith(".ut")
conf.temp_files.pop()

get_temp_file(True).startswith("/tmp/scapy") and len(conf.temp_files) == 0

sane("A\x00\xFFB") == "A..B"

linehexdump(Ether(), dump=True) == "FFFFFFFFFFFF0242D077E8129000 .......B.w...."

chexdump(Ether(), dump=True) == "0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x02, 0x42, 0xd0, 0x77, 0xe8, 0x12, 0x90, 0x00"

hexstr("A\x00\xFFB") == "41 00 ff 42  A..B"

fletcher16_checksum("\x28\x07") == 22319

tex_escape("$#_") == "\\$\\#\\_"

f = colgen(range(3))
len([f.next() for i in range(2)]) == 2

f = incremental_label()
[f.next() for i in range(2)] == ["tag00000", "tag00001"]

import random
random.seed(0x2807)
corrupt_bytes("ABCDE") == "ABCDW"
sane(corrupt_bytes("ABCDE", n=3)) == "A.8D4"

corrupt_bits("ABCDE") == "EBCDE"
sane(corrupt_bits("ABCDE", n=3)) == "AF.EE"

= Test utility functions - network related
~ netaccess

atol("www.secdev.org") == 3642339845


############
############
+ Basic tests

* Those test are here mainly to check nothing has been broken
* and to catch Exceptions

= Building some packets
~ basic IP TCP UDP NTP LLC SNAP Dot11
IP()/TCP()
Ether()/IP()/UDP()/NTP()
Dot11()/LLC()/SNAP()/IP()/TCP()/"XXX"
IP(ttl=25)/TCP(sport=12, dport=42)
IP().summary()

= Manipulating some packets
~ basic IP TCP
a=IP(ttl=4)/TCP()
a.ttl
a.ttl=10
del(a.ttl)
a.ttl
TCP in a
a[TCP]
a[TCP].dport=[80,443]
a
assert(a.copy().time == a.time)
a=3


= Checking overloads
~ basic IP TCP Ether
a=Ether()/IP()/TCP()
a.proto
_ == 6


= sprintf() function
~ basic sprintf Ether IP UDP NTP
a=Ether()/IP()/IP(ttl=4)/UDP()/NTP()
a.sprintf("%type% %IP.ttl% %#05xr,UDP.sport% %IP:2.ttl%")
_ in [ '0x800 64 0x07b 4', 'IPv4 64 0x07b 4']


= sprintf() function 
~ basic sprintf IP TCP SNAP LLC Dot11
* This test is on the conditionnal substring feature of <tt>sprintf()</tt>
a=Dot11()/LLC()/SNAP()/IP()/TCP()
a.sprintf("{IP:{TCP:flags=%TCP.flags%}{UDP:port=%UDP.ports%} %IP.src%}")
_ == 'flags=S 127.0.0.1'


= haslayer function
~ basic haslayer IP TCP ICMP ISAKMP
x=IP(id=1)/ISAKMP_payload_SA(prop=ISAKMP_payload_SA(prop=IP()/ICMP()))/TCP()
TCP in x, ICMP in x, IP in x, UDP in x
_ == (True,True,True,False)

= getlayer function
~ basic getlayer IP ISAKMP UDP
x=IP(id=1)/ISAKMP_payload_SA(prop=IP(id=2)/UDP(dport=1))/IP(id=3)/UDP(dport=2)
x[IP]
x[IP:2]
x[IP:3]
x.getlayer(IP,3)
x.getlayer(IP,4)
x[UDP]
x[UDP:1]
x[UDP:2]
assert(x[IP].id == 1 and x[IP:2].id == 2 and x[IP:3].id == 3 and 
       x.getlayer(IP).id == 1 and x.getlayer(IP,3).id == 3 and
       x.getlayer(IP,4) == None and
       x[UDP].dport == 1 and x[UDP:2].dport == 2)
try:
    x[IP:4]
except IndexError:
    True
else:
    False

= equality
~ basic
w=Ether()/IP()/UDP(dport=53)
x=Ether()/IP(version=4)/UDP()
y=Ether()/IP()/UDP(dport=4)
z=Ether()/IP()/UDP()/NTP()
t=Ether()/IP()/TCP()
x==y, x==z, x==t, y==z, y==t, z==t, w==x
_ == (False, False, False, False, False, False, True)

= answers
~ basic
a1, a2 = "1.2.3.4", "5.6.7.8"
p1 = IP(src=a1, dst=a2)/ICMP(type=8)
p2 = IP(src=a2, dst=a1)/ICMP(type=0)
assert p1.hashret() == p2.hashret()
assert not p1.answers(p2)
assert p2.answers(p1)
conf_back = conf.checkIPinIP
conf.checkIPinIP = True
px = [IP()/p1, IPv6()/p1]
assert not any(p.hashret() == p2.hashret() for p in px)
assert not any(p.answers(p2) for p in px)
assert not any(p2.answers(p) for p in px)
conf.checkIPinIP = False
assert all(p.hashret() == p2.hashret() for p in px)
assert not any(p.answers(p2) for p in px)
assert all(p2.answers(p) for p in px)
conf.checkIPinIP = conf_back


############
############
+ Tests on padding

= Padding assembly
str(Padding("abc"))
assert( _ == "abc" )
str(Padding("abc")/Padding("def"))
assert( _ == "abcdef" )
str(Raw("ABC")/Padding("abc")/Padding("def"))
assert( _ == "ABCabcdef" )
str(Raw("ABC")/Padding("abc")/Raw("DEF")/Padding("def"))
assert( _ == "ABCDEFabcdef" )

= Padding and length computation
IP(str(IP()/Padding("abc")))
assert( _.len == 20 and len(_) == 23 )
IP(str(IP()/Raw("ABC")/Padding("abc")))
assert( _.len == 23 and len(_) == 26 )
IP(str(IP()/Raw("ABC")/Padding("abc")/Padding("def")))
assert( _.len == 23 and len(_) == 29 )

= PadField test
~ PadField padding

class TestPad(Packet):
    fields_desc = [ PadField(StrNullField("st", ""),4), StrField("id", "")]

TestPad() == TestPad(str(TestPad()))


############
############
+ Tests on default value changes mechanism

= Creation of an IPv3 class from IP class with different default values
class IPv3(IP):
    version = 3
    ttl = 32

= Test of IPv3 class
a = IPv3()
a.version, a.ttl
assert(_ == (3,32))
str(a)
assert(_ == '5\x00\x00\x14\x00\x01\x00\x00 \x00\xac\xe7\x7f\x00\x00\x01\x7f\x00\x00\x01')


############
############
+ ISAKMP transforms test

= ISAKMP creation
~ IP UDP ISAKMP 
p=IP(src='192.168.8.14',dst='10.0.0.1')/UDP()/ISAKMP()/ISAKMP_payload_SA(prop=ISAKMP_payload_Proposal(trans=ISAKMP_payload_Transform(transforms=[('Encryption', 'AES-CBC'), ('Hash', 'MD5'), ('Authentication', 'PSK'), ('GroupDesc', '1536MODPgr'), ('KeyLength', 256), ('LifeType', 'Seconds'), ('LifeDuration', 86400L)])/ISAKMP_payload_Transform(res2=12345,transforms=[('Encryption', '3DES-CBC'), ('Hash', 'SHA'), ('Authentication', 'PSK'), ('GroupDesc', '1024MODPgr'), ('LifeType', 'Seconds'), ('LifeDuration', 86400L)])))
p.show()
p


= ISAKMP manipulation
~ ISAKMP
p[ISAKMP_payload_Transform:2]
_.res2 == 12345

= ISAKMP assembly
~ ISAKMP 
hexdump(p)
str(p) == "E\x00\x00\x96\x00\x01\x00\x00@\x11\xa7\x9f\xc0\xa8\x08\x0e\n\x00\x00\x01\x01\xf4\x01\xf4\x00\x82\xbf\x1e\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x10\x00\x00\x00\x00\x00\x00\x00\x00\x00z\x00\x00\x00^\x00\x00\x00\x01\x00\x00\x00\x01\x00\x00\x00R\x01\x01\x00\x00\x03\x00\x00'\x00\x01\x00\x00\x80\x01\x00\x07\x80\x02\x00\x01\x80\x03\x00\x01\x80\x04\x00\x05\x80\x0e\x01\x00\x80\x0b\x00\x01\x00\x0c\x00\x03\x01Q\x80\x00\x00\x00#\x00\x0109\x80\x01\x00\x05\x80\x02\x00\x02\x80\x03\x00\x01\x80\x04\x00\x02\x80\x0b\x00\x01\x00\x0c\x00\x03\x01Q\x80"


= ISAKMP disassembly
~ ISAKMP
q=IP(str(p))
q.show()
q[ISAKMP_payload_Transform:2]
_.res2 == 12345


############
############
+ TFTP tests

= TFTP Options
x=IP()/UDP(sport=12345)/TFTP()/TFTP_RRQ(filename="fname")/TFTP_Options(options=[TFTP_Option(oname="blksize", value="8192"),TFTP_Option(oname="other", value="othervalue")])
assert( str(x) == 'E\x00\x00H\x00\x01\x00\x00@\x11|\xa2\x7f\x00\x00\x01\x7f\x00\x00\x0109\x00E\x004B6\x00\x01fname\x00octet\x00blksize\x008192\x00other\x00othervalue\x00' )
y=IP(str(x))
y[TFTP_Option].oname
y[TFTP_Option:2].oname
assert(len(y[TFTP_Options].options) == 2 and y[TFTP_Option].oname == "blksize")


############
############
+ Dot11 tests


= WEP tests
~ wifi crypto Dot11 LLC SNAP IP TCP
conf.wepkey = "Fobar"
str(Dot11WEP()/LLC()/SNAP()/IP()/TCP(seq=12345678))
assert(_ == '\x00\x00\x00\x00\xe3OjYLw\xc3x_%\xd0\xcf\xdeu-\xc3pH#\x1eK\xae\xf5\xde\xe7\xb8\x1d,\xa1\xfe\xe83\xca\xe1\xfe\xbd\xfe\xec\x00)T`\xde.\x93Td\x95C\x0f\x07\xdd')
Dot11WEP(_)
assert(TCP in _ and _[TCP].seq == 12345678)


############
############
+ SNMP tests

= SNMP assembling
~ SNMP ASN1
str(SNMP())
assert(_ == '0\x18\x02\x01\x01\x04\x06public\xa0\x0b\x02\x01\x00\x02\x01\x00\x02\x01\x000\x00')
SNMP(version="v2c", community="ABC", PDU=SNMPbulk(id=4,varbindlist=[SNMPvarbind(oid="1.2.3.4",value=ASN1_INTEGER(7)),SNMPvarbind(oid="4.3.2.1.2.3",value=ASN1_IA5_STRING("testing123"))]))
str(_)
assert(_ == '05\x02\x01\x01\x04\x03ABC\xa5+\x02\x01\x04\x02\x01\x00\x02\x01\x000 0\x08\x06\x03*\x03\x04\x02\x01\x070\x14\x06\x06\x81#\x02\x01\x02\x03\x16\ntesting123')

= SNMP disassembling
~ SNMP ASN1
x=SNMP('0y\x02\x01\x00\x04\x06public\xa2l\x02\x01)\x02\x01\x00\x02\x01\x000a0!\x06\x12+\x06\x01\x04\x01\x81}\x08@\x04\x02\x01\x07\n\x86\xde\xb78\x04\x0b172.31.19.20#\x06\x12+\x06\x01\x04\x01\x81}\x08@\x04\x02\x01\x07\n\x86\xde\xb76\x04\r255.255.255.00\x17\x06\x12+\x06\x01\x04\x01\x81}\x08@\x04\x02\x01\x05\n\x86\xde\xb9`\x02\x01\x01')
x.show()
assert(x.community=="public" and x.version == 0)
assert(x.PDU.id == 41 and len(x.PDU.varbindlist) == 3)
assert(x.PDU.varbindlist[0].oid == "1.3.6.1.4.1.253.8.64.4.2.1.7.10.14130104")
assert(x.PDU.varbindlist[0].value == "172.31.19.2")
assert(x.PDU.varbindlist[2].oid == "1.3.6.1.4.1.253.8.64.4.2.1.5.10.14130400")
assert(x.PDU.varbindlist[2].value == 1)


############
############
+ Network tests

* Those tests need network access

= Sending and receiving an ICMP
~ netaccess IP ICMP
x=sr1(IP(dst="www.google.com")/ICMP(),timeout=3)
x
assert x[IP].ottl() in [32, 64, 128, 255]
assert 0 <= x[IP].hops() <= 126
x is not None and ICMP in x and x[ICMP].type == 0

= DNS request
~ netaccess IP UDP DNS
* A possible cause of failure could be that the open DNS (resolver1.opendns.com)
* is not reachable or down.
dns_ans = sr1(IP(dst="resolver1.opendns.com")/UDP()/DNS(rd=1,qd=DNSQR(qname="www.slashdot.com")),timeout=5)
DNS in dns_ans

= Whois request
~ netaccess IP
* This test retries on failure because it often fails
import time
import socket
success = False
for i in xrange(5):
    try:
        IP(src="8.8.8.8").whois()
    except socket.error:
        time.sleep(2)
    else:
        success = True
        break

assert success

= AS resolvers
~ netaccess IP
* This test retries on failure because it often fails

success = False
for i in xrange(5):
    try:
        ret = conf.AS_resolver.resolve("8.8.8.8", "8.8.4.4")
    except socket.error:
        time.sleep(2)
    else:
        success = True
        break

assert (len(ret) == 2)

all(x[1] == 15169 for x in ret)


############
############
+ More complex tests

= Implicit logic
~ IP TCP
a=IP(ttl=(5,10))/TCP(dport=[80,443])
ls(a)
[p for p in a]
len(_) == 12


############
############
+ Real usages

= Port scan
~ netaccess IP TCP
ans,unans=sr(IP(dst="www.google.com/30")/TCP(dport=[80,443]),timeout=2)
ans.make_table(lambda (s,r): (s.dst, s.dport, r.sprintf("{TCP:%TCP.flags%}{ICMP:%ICMP.code%}")))

= Traceroute function
~ netaccess
* Let's test traceroute
traceroute("www.slashdot.org")
ans,unans=_

= Result manipulation
~ netaccess
ans.nsummary()
s,r=ans[0]
s.show()
s.show(2)

= DNS packet manipulation
~ netaccess IP UDP DNS
dns_ans.show()
dns_ans.show2()
dns_ans[DNS].an.show()
dns_ans2 = IP(str(dns_ans))
DNS in dns_ans2
assert(str(dns_ans2) == str(dns_ans))
dns_ans2.qd.qname = "www.secdev.org."
* We need to recalculate these values
del(dns_ans2[IP].len)
del(dns_ans2[IP].chksum)
del(dns_ans2[UDP].len)
del(dns_ans2[UDP].chksum)
assert("\x03www\x06secdev\x03org\x00" in str(dns_ans2))
assert(DNS in IP(str(dns_ans2)))

= Arping
~ netaccess
* This test assumes the local network is a /24. This is bad.
conf.route.route("0.0.0.0")[2]
arping(_+"/24")

= send() and sniff()
~ netaccess
import time
import os
def _send_or_sniff(pkt, timeout, flt, pid, fork, t_other=None):
    assert pid != -1
    if pid == 0:
        time.sleep(1)
        (sendp if isinstance(pkt, (Ether, Dot3)) else send)(pkt)
        if fork:
            os._exit(0)
        else:
            return
    else:
        spkt = str(pkt)
        pkts = sniff(
            timeout=timeout, filter=flt,
            stop_filter=lambda p: pkt.__class__ in p and str(p[pkt.__class__]) == spkt
        )
        if fork:
            os.waitpid(pid, 0)
        else:
            t_other.join()
    assert str(pkt) in (str(p[pkt.__class__]) for p in pkts if pkt.__class__ in p)

def send_and_sniff(pkt, timeout=2, flt=None):
    """Send a packet, sniff, and check the packet has been seen"""
    if hasattr(os, "fork"):
        _send_or_sniff(pkt, timeout, flt, os.fork(), True)
    else:
        from threading import Thread
        def run_function(pkt, timeout, flt, pid, thread, results):
            _send_or_sniff(pkt, timeout, flt, pid, False, thread)
            results.put(True)
        results = Queue.Queue()
        t_parent = Thread(target=run_function, args=(pkt, timeout, flt, 0, None, results))
        t_child = Thread(target=run_function, args=(pkt, timeout, flt, 1, t_parent, results))
        t_parent.start()
        t_child.start()
        t_parent.join()
        t_child.join()
        assert results.qsize() >= 2
        while not results.empty():
            assert results.get()

send_and_sniff(IP(dst="secdev.org")/ICMP())
send_and_sniff(IP(dst="secdev.org")/ICMP(), flt="icmp")
send_and_sniff(Ether()/IP(dst="secdev.org")/ICMP())


############
############
+ Automaton tests

= Simple automaton
~ automaton
class ATMT1(Automaton):
    def parse_args(self, init, *args, **kargs):
        Automaton.parse_args(self, *args, **kargs)
        self.init = init
    @ATMT.state(initial=1)
    def BEGIN(self):
      	raise self.MAIN(self.init)
    @ATMT.state()
    def MAIN(self, s):
        return s
    @ATMT.condition(MAIN, prio=-1)
    def go_to_END(self, s):
        if len(s) > 20:
            raise self.END(s).action_parameters(s)
    @ATMT.condition(MAIN)
    def trA(self, s):
        if s.endswith("b"):
            raise self.MAIN(s+"a")
    @ATMT.condition(MAIN)
    def trB(self, s):
        if s.endswith("a"):
            raise self.MAIN(s*2+"b")
    @ATMT.state(final=1)
    def END(self, s):
        return s
    @ATMT.action(go_to_END)
    def action_test(self, s):
        self.result = s
    
= Simple automaton Tests
~ automaton

a=ATMT1(init="a", ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'aabaaababaaabaaababab' )
a.result
assert( _ == 'aabaaababaaabaaababab' )
a=ATMT1(init="b", ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'babababababababababababababab' )
a.result
assert( _ == 'babababababababababababababab' )

= Simple automaton stuck test
~ automaton

try:    
    ATMT1(init="", ll=lambda: None, recvsock=lambda: None).run()
except Automaton.Stuck:
    True
else:
    False


= Automaton state overloading
~ automaton
class ATMT2(ATMT1):
    @ATMT.state()
    def MAIN(self, s):
        return "c"+ATMT1.MAIN(self, s).run()

a=ATMT2(init="a", ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'ccccccacabacccacababacccccacabacccacababab' )


a.result
assert( _ == 'ccccccacabacccacababacccccacabacccacababab' )
a=ATMT2(init="b", ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'cccccbaccbabaccccbaccbabab')
a.result
assert( _ == 'cccccbaccbabaccccbaccbabab')


= Automaton condition overloading
~ automaton
class ATMT3(ATMT2):
    @ATMT.condition(ATMT1.MAIN)
    def trA(self, s):
        if s.endswith("b"):
            raise self.MAIN(s+"da")


a=ATMT3(init="a", debug=2, ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'cccccacabdacccacabdabda')
a.result
assert( _ == 'cccccacabdacccacabdabda')
a=ATMT3(init="b", ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'cccccbdaccbdabdaccccbdaccbdabdab' )

a.result
assert( _ == 'cccccbdaccbdabdaccccbdaccbdabdab' )


= Automaton action overloading
~ automaton
class ATMT4(ATMT3):
    @ATMT.action(ATMT1.go_to_END)
    def action_test(self, s):
        self.result = "e"+s+"e"

a=ATMT4(init="a", ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'cccccacabdacccacabdabda')
a.result
assert( _ == 'ecccccacabdacccacabdabdae')
a=ATMT4(init="b", ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'cccccbdaccbdabdaccccbdaccbdabdab' )
a.result
assert( _ == 'ecccccbdaccbdabdaccccbdaccbdabdabe' )


= Automaton priorities
~ automaton
class ATMT5(Automaton):
    @ATMT.state(initial=1)
    def BEGIN(self):
        self.res = "J"
    @ATMT.condition(BEGIN, prio=1)
    def tr1(self):
        self.res += "i"
        raise self.END()
    @ATMT.condition(BEGIN)
    def tr2(self):
        self.res += "p"
    @ATMT.condition(BEGIN, prio=-1)
    def tr3(self):
        self.res += "u"
    
    @ATMT.action(tr1)
    def ac1(self):
        self.res += "e"
    @ATMT.action(tr1, prio=-1)
    def ac2(self):
        self.res += "t"
    @ATMT.action(tr1, prio=1)
    def ac3(self):
        self.res += "r"
   
    @ATMT.state(final=1)
    def END(self):
        return self.res

a=ATMT5(ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'Jupiter' )

= Automaton test same action for many conditions
~ automaton
class ATMT6(Automaton):
    @ATMT.state(initial=1)
    def BEGIN(self):
        self.res="M"
    @ATMT.condition(BEGIN)
    def tr1(self):
        raise self.MIDDLE()
    @ATMT.action(tr1) # default prio=0
    def add_e(self):
        self.res += "e"
    @ATMT.action(tr1, prio=2)
    def add_c(self):
        self.res += "c"
    @ATMT.state()
    def MIDDLE(self):
        self.res += "u"
    @ATMT.condition(MIDDLE)
    def tr2(self):
        raise self.END()
    @ATMT.action(tr2, prio=2)
    def add_y(self):
        self.res += "y"
    @ATMT.action(tr1, prio=1)
    @ATMT.action(tr2)
    def add_r(self):
        self.res += "r"
    @ATMT.state(final=1)
    def END(self):
        return self.res

a=ATMT6(ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == 'Mercury' )

a.restart()
a.run()
assert( _ == 'Mercury' )

= Automaton test io event
~ automaton

class ATMT7(Automaton):
    @ATMT.state(initial=1)
    def BEGIN(self):
        self.res = "S"
    @ATMT.ioevent(BEGIN, name="tst")
    def tr1(self, fd):
        self.res += fd.recv()
        raise self.NEXT_STATE()
    @ATMT.state()
    def NEXT_STATE(self):
        self.oi.tst.send("ur")
    @ATMT.ioevent(NEXT_STATE, name="tst")
    def tr2(self, fd):
        self.res += fd.recv()
        raise self.END()
    @ATMT.state(final=1)
    def END(self):
        self.res += "n"
        return self.res

a=ATMT7(ll=lambda: None, recvsock=lambda: None)
a.run(wait=False)
a.io.tst.send("at")
a.io.tst.recv()
a.io.tst.send(_)
a.run()
assert( _ == "Saturn" )

a.restart()
a.run(wait=False)
a.io.tst.send("at")
a.io.tst.recv()
a.io.tst.send(_)
a.run()
assert( _ == "Saturn" )

= Automaton test io event from external fd
~ automaton
import os

class ATMT8(Automaton):
    @ATMT.state(initial=1)
    def BEGIN(self):
        self.res = "U"
    @ATMT.ioevent(BEGIN, name="extfd")
    def tr1(self, fd):
        self.res += fd.read(2)
        raise self.NEXT_STATE()
    @ATMT.state()
    def NEXT_STATE(self):
        pass
    @ATMT.ioevent(NEXT_STATE, name="extfd")
    def tr2(self, fd):
        self.res += fd.read(2)
        raise self.END()
    @ATMT.state(final=1)
    def END(self):
        self.res += "s"
        return self.res

if WINDOWS:
    r = w = ObjectPipe()
else:
    r,w = os.pipe()

def writeOn(w, msg):
    if WINDOWS:
        w.write(msg)
    else:
        os.write(w, msg)

a=ATMT8(external_fd={"extfd":r}, ll=lambda: None, recvsock=lambda: None)
a.run(wait=False)
writeOn(w,"ra")
writeOn(w,"nu")

a.run()
assert( _ == "Uranus" )

a.restart()
a.run(wait=False)
writeOn(w,"ra")
writeOn(w,"nu")
a.run()
assert( _ == "Uranus" )

= Automaton test interception_points, and restart
~ automaton
class ATMT9(Automaton):
    def my_send(self, x):
        self.io.loop.send(x)
    @ATMT.state(initial=1)
    def BEGIN(self):
        self.res = "V"
        self.send(Raw("ENU"))
    @ATMT.ioevent(BEGIN, name="loop")
    def received_sth(self, fd):
        self.res += fd.recv().load
        raise self.END()
    @ATMT.state(final=1)
    def END(self):
        self.res += "s"
        return self.res

a=ATMT9(debug=5, ll=lambda: None, recvsock=lambda: None)
a.run()
assert( _ == "VENUs" )

a.restart()
a.run()
assert( _ == "VENUs" )

a.restart()
a.BEGIN.intercepts()
while True:
    try:
        x = a.run()
    except Automaton.InterceptionPoint,p:
        a.accept_packet(Raw(p.packet.load.lower()), wait=False)
    else:
        break

x
assert( _ == "Venus" )


############
############
+ Test IP options

= IP options individual assembly
~ IP options
str(IPOption())
assert(_ == '\x00\x02')
str(IPOption_NOP())
assert(_ == '\x01')
str(IPOption_EOL())
assert(_ == '\x00')
str(IPOption_LSRR(routers=["1.2.3.4","5.6.7.8"]))
assert(_ == '\x83\x0b\x04\x01\x02\x03\x04\x05\x06\x07\x08')

= IP options individual dissection
~ IP options
IPOption("\x00")
assert(_.option == 0 and isinstance(_, IPOption_EOL))
IPOption("\x01")
assert(_.option == 1 and isinstance(_, IPOption_NOP))
lsrr='\x83\x0b\x04\x01\x02\x03\x04\x05\x06\x07\x08'
p=IPOption_LSRR(lsrr)
p
q=IPOption(lsrr)
q
assert(p == q)

= IP assembly and dissection with options
~ IP options
p = IP(src="9.10.11.12",dst="13.14.15.16",options=IPOption_SDBM(addresses=["1.2.3.4","5.6.7.8"]))/TCP()
str(p)
assert(_ == 'H\x00\x004\x00\x01\x00\x00@\x06\xa2q\t\n\x0b\x0c\r\x0e\x0f\x10\x95\n\x01\x02\x03\x04\x05\x06\x07\x08\x00\x00\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00_K\x00\x00')
q=IP(_)
q
assert( isinstance(q.options[0],IPOption_SDBM) )
assert( q[IPOption_SDBM].addresses[1] == "5.6.7.8" )
p.options[0].addresses[0] = '5.6.7.8'
assert( IP(str(p)).options[0].addresses[0] == '5.6.7.8' )
IP(src="9.10.11.12", dst="13.14.15.16", options=[IPOption_NOP(),IPOption_LSRR(routers=["1.2.3.4","5.6.7.8"]),IPOption_Security(transmission_control_code="XYZ")])/TCP()
str(_)
assert(_ == 'K\x00\x00@\x00\x01\x00\x00@\x06\xf3\x83\t\n\x0b\x0c\r\x0e\x0f\x10\x01\x83\x0b\x04\x01\x02\x03\x04\x05\x06\x07\x08\x82\x0b\x00\x00\x00\x00\x00\x00XYZ\x00\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00_K\x00\x00')
IP(_)
q=_
assert(q[IPOption_LSRR].get_current_router() == "1.2.3.4")
assert(q[IPOption_Security].transmission_control_code == "XYZ")
assert(q[TCP].flags == 2)


############
############
+ Test PPP

= PPP/HDLC
~ ppp hdlc
HDLC()/PPP()/PPP_IPCP()
str(_)
s=_
assert(s == '\xff\x03\x80!\x01\x00\x00\x04')
PPP(s)
p=_
assert(HDLC in p)
assert(p[HDLC].control==3)
assert(p[PPP].proto==0x8021)
PPP(s[2:])
q=_
assert(HDLC not in q)
assert(q[PPP].proto==0x8021)


= PPP IPCP
~ ppp ipcp
PPP('\x80!\x01\x01\x00\x10\x03\x06\xc0\xa8\x01\x01\x02\x06\x00-\x0f\x01')