isotp vecu echo test

Cargo.toml

@@ -13,6 +13,7 @@ repository = "https://github.com/I-CAN-hack/automotive"
 all = []
 test_panda = []
 test_socketcan = []
+test_vcan = []
 serde = ["dep:serde"]
 
 [dependencies]

scripts/set_up_vcan.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env sh
+
+sudo modprobe vcan
+sudo ip link add dev vcan0 type vcan
+sudo ip link set dev vcan0 up

scripts/vecu.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+import argparse
+import threading
+import time
+
+from scapy.all import *
+from scapy.ansmachine import AnsweringMachine
+
+
+def forwarding1(pkt):
+    return pkt
+
+def forwarding2(pkt):
+    return False
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--iface", type=str, default="vcan0")
+    parser.add_argument("--rx", type=int, default=0x7a1)
+    parser.add_argument("--tx", type=int, default=0x7a9)
+    parser.add_argument("--protocol", type=str, default="iso-tp")
+    parser.add_argument("--timeout", type=int, default=10)
+
+    args = parser.parse_args()
+
+    conf.contribs['ISOTP'] = {'use-can-isotp-kernel-module': False}
+    load_contrib('isotp')
+    load_contrib('automotive.uds')
+    load_contrib('automotive.ecu')
+
+
+    if args.protocol == 'uds':
+        with ISOTPSocket(args.iface, tx_id=args.tx, rx_id=args.rx, basecls=UDS) as sock:
+            resp = [] # TODO: Add responses
+            ecu = EcuAnsweringMachine(supported_responses=resp, main_socket=sock, basecls=UDS)
+            sim = threading.Thread(target=ecu, kwargs={'count': 4, 'timeout': args.timeout})
+            sim.start()
+
+    elif args.protocol == 'iso-tp':
+        with ISOTPSocket(args.iface, tx_id=args.tx, rx_id=args.rx) as sock1:
+            with ISOTPSocket(args.iface, tx_id=args.tx, rx_id=args.rx) as sock2:
+                sock1.send(b'\xAA') # Signal to test that ECU is ready
+                bridge_and_sniff(if1=sock1, if2=sock2, xfrm12=forwarding1, xfrm21=forwarding2, timeout=args.timeout)

tests/isotp_tests.rs

@@ -0,0 +1,50 @@
+use automotive::async_can::AsyncCanAdapter;
+use automotive::can::Identifier;
+use automotive::isotp::{IsoTPAdapter, IsoTPConfig};
+use tokio_stream::StreamExt;
+use std::process::Command;
+use std::vec;
+
+static VECU_STARTUP_TIMEOUT_MS: u64 = 1000;
+
+async fn vecu_spawn(adapter: &AsyncCanAdapter) -> std::process::Child {
+    let stream = adapter.recv().timeout(std::time::Duration::from_millis(VECU_STARTUP_TIMEOUT_MS));
+    tokio::pin!(stream);
+
+    let vecu = Command::new("scripts/vecu.py").spawn().unwrap();
+    stream.next().await.unwrap().expect("vecu did not start");
+
+    vecu
+}
+
+
+async fn isotp_echo(msg_len: usize) {
+    let adapter = automotive::socketcan::SocketCan::new_async_from_name("vcan0").unwrap();
+    let mut vecu = vecu_spawn(&adapter).await;
+
+    let config = IsoTPConfig::new(0, Identifier::Standard(0x7a1));
+    let isotp = IsoTPAdapter::new(&adapter, config);
+
+    let mut stream = isotp.recv();
+    let request = vec![0xaa; msg_len];
+    isotp.send(&request).await.unwrap();
+    let response = stream.next().await.unwrap().unwrap();
+
+    assert_eq!(response, request);
+
+    vecu.kill().unwrap();
+
+}
+
+// #[cfg(feature = "test_vcan")]
+#[tokio::test]
+#[serial_test::serial]
+async fn isotp_test_single_frame() {
+    isotp_echo(7).await;
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn isotp_test_flow_control() {
+    isotp_echo(64).await;
+}