GitHub Viewer
(*
* Copyright 2025 Multikernel Technologies, Inc.
*
* 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.
*)
open Kernelscript.Parse
open Kernelscript.Type_checker
open Alcotest
(** Test program reference type checking *)
let test_program_reference_type () =
let program_text = {|
@xdp fn packet_filter(ctx: *xdp_md) -> xdp_action {
return 2
}
fn main() -> i32 {
var prog_handle = load(packet_filter)
var result = attach(prog_handle, "eth0", 0)
return 0
}
|} in
try
let ast = parse_string program_text in
let _ = Kernelscript.Symbol_table.build_symbol_table ast in
let (typed_ast, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ast in
check bool "program reference type checking" true (List.length typed_ast > 0)
with
| e -> fail ("program reference type checking failed: " ^ Printexc.to_string e)
(** Test program reference with different program types *)
let test_different_program_types () =
let program_text = {|
@probe("sys_read") fn kprobe_tracer(fd: u32, buf: *u8, count: size_t) -> i32 {
return 0
}
@tc("ingress") fn tc_filter(ctx: *__sk_buff) -> i32 {
return 0
}
fn main() -> i32 {
var kprobe_handle = load(kprobe_tracer)
var tc_handle = load(tc_filter)
var kprobe_result = attach(kprobe_handle, "sys_read", 0)
var tc_result = attach(tc_handle, "eth0", 1)
return 0
}
|} in
try
let ast = parse_string program_text in
let _ = Kernelscript.Symbol_table.build_symbol_table ast in
let (typed_ast, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ast in
check bool "different program types" true (List.length typed_ast > 0)
with
| e -> fail ("different program types failed: " ^ Printexc.to_string e)
(** Test invalid program reference *)
let test_invalid_program_reference () =
let program_text = {|
fn main() -> i32 {
var prog_handle = load(non_existent_program)
return 0
}
|} in
try
let ast = parse_string program_text in
let _ = Kernelscript.Symbol_table.build_symbol_table ast in
let (_, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ast in
fail "should fail for non-existent program"
with
| Type_error _ -> ()
| Kernelscript.Symbol_table.Symbol_error _ -> ()
| e -> fail ("Expected Type_error or Symbol_error, got: " ^ Printexc.to_string e)
(** Test program reference as variable *)
let test_program_reference_as_variable () =
let program_text = {|
@xdp fn my_xdp(ctx: *xdp_md) -> xdp_action {
return 2
}
fn main() -> i32 {
var prog_ref = my_xdp // Should work - program reference as variable
var prog_handle = load(prog_ref)
return 0
}
|} in
try
let ast = parse_string program_text in
let _ = Kernelscript.Symbol_table.build_symbol_table ast in
let (typed_ast, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ast in
check bool "program reference as variable" true (List.length typed_ast > 0)
with
| e -> fail ("program reference as variable failed: " ^ Printexc.to_string e)
(** Test wrong argument types for program functions *)
let test_wrong_argument_types () =
let program_text = {|
@xdp fn my_xdp(ctx: *xdp_md) -> xdp_action {
return 2
}
fn main() -> i32 {
var prog_handle = load("string_instead_of_program") // Should fail
return 0
}
|} in
try
let ast = parse_string program_text in
let _ = Kernelscript.Symbol_table.build_symbol_table ast in
let (_, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ast in
fail "should fail for wrong argument type"
with
| Type_error _ -> ()
| e -> fail ("Expected Type_error, got: " ^ Printexc.to_string e)
(** Test stdlib integration *)
let test_stdlib_integration () =
(* Test that the built-in functions are properly recognized *)
check bool "load is builtin" true (Kernelscript.Stdlib.is_builtin_function "load");
check bool "attach is builtin" true (Kernelscript.Stdlib.is_builtin_function "attach");
(* Test getting function signatures *)
(match Kernelscript.Stdlib.get_builtin_function_signature "load" with
| Some (params, return_type) ->
check int "load parameter count" 1 (List.length params);
check bool "load return type is ProgramHandle" true (return_type = Kernelscript.Ast.ProgramHandle)
| None -> check bool "load function signature should exist" false true);
(match Kernelscript.Stdlib.get_builtin_function_signature "attach" with
| Some (params, return_type) ->
(* attach uses custom validation (param_types = []), so count is 0 *)
check int "attach parameter count" 0 (List.length params);
check bool "attach return type is U32" true (return_type = Kernelscript.Ast.U32)
| None -> check bool "attach function signature should exist" false true);
(match Kernelscript.Stdlib.get_builtin_function_signature "read" with
| Some (params, return_type) ->
check int "read parameter count" 0 (List.length params);
check bool "read return type is perf_read" true (return_type = Kernelscript.Ast.Struct "perf_read")
| None -> check bool "read function signature should exist" false true);
(* Verify that the custom validation function is wired up on the attach entry *)
(match Kernelscript.Stdlib.get_builtin_function "attach" with
| Some func ->
check bool "attach has custom validation wired up" true
(match func.validate with Some _ -> true | None -> false)
| None -> check bool "attach builtin should exist" false true)
(** Test perf attach returns an attachment value that can be read/detached. *)
let test_perf_attachment_value_flow () =
let program_text = {|
@perf_event fn on_cache_miss(ctx: *bpf_perf_event_data) -> i32 {
return 0
}
fn main() -> i32 {
var prog = load(on_cache_miss)
var att = attach(prog, perf_options {
event: cache_misses,
period: 1000000,
}, 0)
var snapshot = read(att)
var count = snapshot.scaled
detach(att)
print("count=%lld", count)
return 0
}
|} in
try
let ast = parse_string program_text in
let symbol_table =
Kernelscript.Symbol_table.build_symbol_table
~builtin_asts:[Kernelscript.Stdlib.get_builtin_types ()]
ast
in
let (typed_ast, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ~symbol_table:(Some symbol_table) ast in
check bool "perf attachment value flow should type check" true (List.length typed_ast > 0)
with
| e -> fail ("perf attachment value flow failed: " ^ Printexc.to_string e)
(** Test that calling attach without load fails *)
let test_attach_without_load_fails () =
let program_text = {|
@xdp fn simple_xdp(ctx: *xdp_md) -> xdp_action {
return 2
}
fn main() -> i32 {
var result = attach(simple_xdp, "eth0", 0) // Should fail - program ref instead of handle
return 0
}
|} in
try
let ast = parse_string program_text in
let _ = Kernelscript.Symbol_table.build_symbol_table ast in
let (_, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ast in
check bool "should fail when attach called with program reference" false true
with
| Type_error (msg, _) ->
check bool "should fail with type error" true (String.length msg > 0);
(* Error message is: "attach() requires (handle, target, flags) — ..." *)
check bool "error message starts with attach()" true
(String.length msg >= 8 && String.sub msg 0 8 = "attach()")
| _ ->
check bool "should fail when attach called with program reference" false true
(** Test multiple program handles with proper resource management *)
let test_multiple_program_handles () =
let program_text = {|
@xdp fn xdp_filter(ctx: *xdp_md) -> xdp_action {
return 2
}
@tc("ingress") fn tc_shaper(ctx: *__sk_buff) -> i32 {
return 0
}
fn main() -> i32 {
var xdp_handle = load(xdp_filter)
var tc_handle = load(tc_shaper)
var xdp_result = attach(xdp_handle, "eth0", 0)
var tc_result = attach(tc_handle, "eth0", 1)
return 0
}
|} in
try
let ast = parse_string program_text in
let _ = Kernelscript.Symbol_table.build_symbol_table ast in
let (typed_ast, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ast in
check bool "multiple program handles should work" true (List.length typed_ast > 0)
with
| e -> fail ("multiple program handles failed: " ^ Printexc.to_string e)
(** Test that program handle variables can be named appropriately *)
let test_program_handle_naming () =
let program_text = {|
@xdp fn simple_xdp(ctx: *xdp_md) -> xdp_action {
return 2
}
fn main() -> i32 {
var program_handle = load(simple_xdp) // Clear, non-fd naming
var network_prog = load(simple_xdp) // Alternative naming
var result1 = attach(program_handle, "eth0", 0)
var result2 = attach(network_prog, "lo", 0)
return 0
}
|} in
try
let ast = parse_string program_text in
let _ = Kernelscript.Symbol_table.build_symbol_table ast in
let (typed_ast, _) = Kernelscript.Type_checker.type_check_and_annotate_ast ast in
check bool "program handle naming should work" true (List.length typed_ast > 0)
with
| e -> fail ("program handle naming failed: " ^ Printexc.to_string e)
(** Test suite *)
let program_ref_tests = [
"program_reference_type_checking", `Quick, test_program_reference_type;
"different_program_types", `Quick, test_different_program_types;
"invalid_program_reference", `Quick, test_invalid_program_reference;
"program_reference_as_variable", `Quick, test_program_reference_as_variable;
"wrong_argument_types", `Quick, test_wrong_argument_types;
"stdlib_integration", `Quick, test_stdlib_integration;
"attach_without_load_fails", `Quick, test_attach_without_load_fails;
"multiple_program_handles", `Quick, test_multiple_program_handles;
"program_handle_naming", `Quick, test_program_handle_naming;
"perf_attachment_value_flow", `Quick, test_perf_attachment_value_flow;
]
let () =
run "Program Reference Tests" [
"program_ref", program_ref_tests;
]