% Regression tests for obdscanner
~ vcan_socket needs_root linux not_pypy automotive_comm scanner

+ Configuration
~ conf

= Imports
with open(scapy_path("test/contrib/automotive/interface_mockup.py")) as f:
    exec(f.read())

load_contrib("automotive.ecu", globals_dict=globals())

+ Usage tests

= Test wrong usage
print(sys.executable)
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
returncode = result.wait()
std_out, std_err = result.communicate()
assert returncode != 0

expected_output = plain_str(b'usage:')
assert expected_output in plain_str(std_err)


= Test show help
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py", "--help"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
assert result.wait() == 0
std_out, std_err = result.communicate()
expected_output = plain_str(b'Scan for all possible obd service classes and their subfunctions.')
assert expected_output in plain_str(std_out)

+ Scan tests

= Load contribution layer

from scapy.contrib.automotive.obd.obd import *

+ Simulate scanner

= Test DTC scan

drain_bus(iface0)

s3 = OBD()/OBD_S03_PR(dtcs=[OBD_DTC()])

example_responses = [EcuResponse(responses=s3)]

with new_can_socket0() as isocan, ISOTPSocket(isocan, 0x7e8, 0x7e0, basecls=OBD, padding=True) as ecu, \
        new_can_socket0() as isocan2, ISOTPSocket(isocan2, 0x7e0, 0x7e8, basecls=OBD, padding=True) as tester:
    conf.verb = -1
    answering_machine = EcuAnsweringMachine(supported_responses=example_responses, main_socket=ecu, basecls=OBD, verbose=False)
    sim = threading.Thread(target=answering_machine, kwargs={'verbose': False, 'timeout': 15, 'stop_filter': lambda p: bytes(p) == b"\x01\xff\xff\xff\xff"})
    sim.start()
    try:
        result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        std_out1, std_err1 = result.communicate()
        result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        std_out2, std_err2 = result.communicate()
    except Exception as e:
        print(e)
    finally:
        tester.send(b"\x01\xff\xff\xff\xff")
        sim.join(timeout=10)
    expected_output = b"1 requests were sent, 1 answered"
    assert bytes_encode(expected_output) in bytes_encode(std_out1) or bytes_encode(expected_output) in bytes_encode(std_out2) 

= Test supported PIDs scan

drain_bus(iface0)

s1_pid00 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID00(supported_pids="PID03+PID0B+PID0F")])
s6_mid00 = OBD()/OBD_S06_PR(data_records=[OBD_S06_PR_Record()/OBD_MID00(supported_mids="")])
s8_tid00 = OBD()/OBD_S08_PR(data_records=[OBD_S08_PR_Record()/OBD_TID00(supported_tids="")])
s9_iid00 = OBD()/OBD_S09_PR(data_records=[OBD_S09_PR_Record()/OBD_IID00(supported_iids="")])

s1_pid03 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID03(fuel_system1=0, fuel_system2=2)])
s1_pid0B = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID0B(data=100)])
s1_pid0F = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID0F(data=50)])

# Create answers for 'supported PIDs scan'
example_responses = \
    [EcuResponse(responses=s3),
     EcuResponse(responses=s1_pid00),
     EcuResponse(responses=s6_mid00),
     EcuResponse(responses=s8_tid00),
     EcuResponse(responses=s9_iid00),
     EcuResponse(responses=s1_pid03),
     EcuResponse(responses=s1_pid0B),
     EcuResponse(responses=s1_pid0F)]



with new_can_socket0() as isocan, ISOTPSocket(isocan, 0x7e8, 0x7e0, basecls=OBD, padding=True) as ecu, \
        new_can_socket0() as isocan2, ISOTPSocket(isocan2, 0x7e0, 0x7e8, basecls=OBD, padding=True) as tester:
    answering_machine = EcuAnsweringMachine(supported_responses=example_responses, main_socket=ecu, basecls=OBD, verbose=False)
    sim = threading.Thread(target=answering_machine, kwargs={'verbose': False, 'timeout': 100, 'stop_filter': lambda p: bytes(p) == b"\x01\xff\xff\xff\xff"})
    sim.start()
    try:
        result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        std_out1, std_err1 = result.communicate()
        print(std_out2, std_err2)
        result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        std_out2, std_err2 = result.communicate()
        print(std_out2, std_err2)
    except Exception as e:
        print(e)
    finally:
        tester.send(b"\x01\xff\xff\xff\xff")
        sim.join(timeout=10)
    expected_output = ["supported_pids=PID0F+PID0B+PID03", "fuel_system1=OpenLoopInsufficientEngineTemperature fuel_system2=ClosedLoop", "data=100 kPa", "data=50.0 deg. C"]
    for out in expected_output:
        assert bytes_encode(out) in bytes_encode(std_out1) or bytes_encode(out) in bytes_encode(std_out2)  

= Test only Service 01 PIDs scan

drain_bus(iface0)

s1_pid00 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID00(supported_pids="PID03+PID0B+PID0F")])
s6_mid00 = OBD()/OBD_S06_PR(data_records=[OBD_S06_PR_Record()/OBD_MID00(supported_mids="")])
s8_tid00 = OBD()/OBD_S08_PR(data_records=[OBD_S08_PR_Record()/OBD_TID00(supported_tids="")])
s9_iid00 = OBD()/OBD_S09_PR(data_records=[OBD_S09_PR_Record()/OBD_IID00(supported_iids="")])

s1_pid03 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID03(fuel_system1=0, fuel_system2=2)])
s1_pid0B = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID0B(data=100)])
s1_pid0F = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID0F(data=50)])

# Create answers for 'supported PIDs scan'
example_responses = \
    [EcuResponse(responses=s3),
     EcuResponse(responses=s1_pid00),
     EcuResponse(responses=s6_mid00),
     EcuResponse(responses=s8_tid00),
     EcuResponse(responses=s9_iid00),
     EcuResponse(responses=s1_pid03),
     EcuResponse(responses=s1_pid0B),
     EcuResponse(responses=s1_pid0F)]



with new_can_socket0() as isocan, ISOTPSocket(isocan, 0x7e8, 0x7e0, basecls=OBD, padding=True) as ecu, \
        new_can_socket0() as isocan2, ISOTPSocket(isocan2, 0x7e0, 0x7e8, basecls=OBD, padding=True) as tester:
    answering_machine = EcuAnsweringMachine(supported_responses=example_responses, main_socket=ecu, basecls=OBD, verbose=False)
    sim = threading.Thread(target=answering_machine, kwargs={'verbose': False, 'timeout': 100, 'stop_filter':  lambda p: bytes(p) == b"\x01\xff\xff\xff\xff"})
    sim.start()
    try:
        result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30", "-1"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        std_out1, std_err1 = result.communicate()
        print(std_out1, std_err1)
        result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30", "-1"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        std_out2, std_err2 = result.communicate()
        print(std_out2, std_err2)
    except Exception as e:
        print(e)
    finally:
        tester.send(b"\x01\xff\xff\xff\xff")
        sim.join(timeout=10)
    expected_output = ["supported_pids=PID0F+PID0B+PID03", "fuel_system1=OpenLoopInsufficientEngineTemperature fuel_system2=ClosedLoop", "data=100 kPa", "data=50.0 deg. C"]
    for out in expected_output:
        assert bytes_encode(out) in bytes_encode(std_out1) or bytes_encode(out) in bytes_encode(std_out2)


= Test full scan

drain_bus(iface0)

# Add unsupported PID
s1_pid01 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID01()])
example_responses.append(EcuResponse(responses=s1_pid01))

with new_can_socket0() as isocan, ISOTPSocket(isocan, 0x7e8, 0x7e0, basecls=OBD, padding=True) as ecu, \
        new_can_socket0() as isocan2, ISOTPSocket(isocan2, 0x7e0, 0x7e8, basecls=OBD, padding=True) as tester:
    answering_machine = EcuAnsweringMachine(supported_responses=example_responses, main_socket=ecu, basecls=OBD, verbose=False)
    sim = threading.Thread(target=answering_machine, kwargs={'verbose': False, 'timeout': 100, 'stop_filter': lambda p: bytes(p) == b"\x01\xff\xff\xff\xff"})
    sim.start()
    try:
        result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30", "-f", "-1", "-3"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        std_out1, std_err1 = result.communicate()
        result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30", "-f", "-1", "-3"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        std_out2, std_err2 = result.communicate()
    except Exception as e:
        print(e)
    finally:
        tester.send(b"\x01\xff\xff\xff\xff")
        sim.join(timeout=10)
    expected_output = ["256 requests were sent", "1 requests were sent, 1 answered"]
    for out in expected_output:
        assert bytes_encode(out) in bytes_encode(std_out1) or bytes_encode(out) in bytes_encode(std_out2)

+ Cleanup

= Delete vcan interfaces

assert cleanup_interfaces()