DEVS2Agents/test/test_transitions.jl

julia_test_file = "test/traces/julia.csv"

@testset "Transient State (ta = 0.0)" begin
    top = CoupledDEVS(:top)

    sender_ref = add_submodel!(top, :sender, MockNode;
        out_ports=(; out=Int), make_out=true, ta=1.0
    )

    transient_node = add_submodel!(top, :transient, MockNode;
        inbox=(; in=Int[]), out_ports=(; out=Int), ta=0.0, forward=true
    )
    connect!(top, sender_ref.out, transient_node.in)

    model = compile_devs(top)
    simulate_devs!(model, 2.0)

    node = model[model.path_to_id["top.transient"]]
    # Proves it received the message, then instantly fired internally without time passing
    @test node.exec_order == [:ta, :external, :ta, :output, :internal, :ta]
    @test abmtime(model) == 1.0 # Global clock should NOT have advanced past the trigger time
end

@testset "Test Order Output-Transition" begin
    top = CoupledDEVS(:top)
    sender_ref = add_submodel!(top, :sender, MockNode;
        out_ports=(; out=Int),
        make_out=true, ta=2.0
    )

    receiver_ref = add_submodel!(top, :receiver, MockNode;
        inbox=(; in=Int[]),
    )

    connect!(top, sender_ref.out, receiver_ref.in)

    model = compile_devs(top)
    simulate_devs!(model)

    sender = model[model.path_to_id["top.sender"]]
    @test sender.n_outputs == 1
    @test sender.n_internal == 1
    @test sender.n_external == 0
    @test sender.n_confluent == 0

    @test sender.exec_order[1:3] == [:ta, :output, :internal]
end

@testset "Obsolete Event Purging" begin
    top = CoupledDEVS(:top)
    # Fast sender, fires every 1.0 seconds
    sender = add_submodel!(top, :sender, MockNode; out_ports=(; out=Int), ta=1.0, cycles=5, make_out=true)

    # Slow receiver, wants to fire at 10.0 seconds
    rec = add_submodel!(top, :rec, MockNode; inbox=(; in=Int[]), ta=4.0)
    connect!(top, sender.out, rec.in)

    model = compile_devs(top)
    simulate_devs!(model, 11.0)

    node = model[model.path_to_id["top.rec"]]
    @test node.n_external == 5
    @test node.n_internal == 1
end

@testset "Internal Transition Test" begin
    top = CoupledDEVS(:top)
    add_submodel!(top, :receiver, MockNode;
        inbox=(; in=Int[]),
        ta=2.0
    )

    model = compile_devs(top; logging=julia_test_file)
    simulate_devs!(model)
    test_trace_equivalence(julia_test_file, "test/traces/Internal.csv")

    receiver = model[model.path_to_id["top.receiver"]]

    @test receiver.n_internal == 1
    @test receiver.n_external == 0
    @test receiver.n_confluent == 0
end

@testset "External Transition Test" begin
    top = CoupledDEVS(:top)

    sender_ref = add_submodel!(top, :sender, MockNode;
        out_ports=(; out=Int),
        ta=1.0, make_out=true
    )
    receiver_ref = add_submodel!(top, :receiver, MockNode;
        inbox=(; in=Int[]),
    )
    connect!(top, sender_ref.out, receiver_ref.in)

    model = compile_devs(top; logging=julia_test_file)
    simulate_devs!(model)
    test_trace_equivalence(julia_test_file, "test/traces/External.csv")

    sender = model[model.path_to_id["top.sender"]]
    @test sender.n_internal == 1
    @test sender.n_external == 0
    @test sender.n_confluent == 0

    receiver = model[model.path_to_id["top.receiver"]]

    @test receiver.n_internal == 0
    @test receiver.n_external == 1
    @test receiver.n_confluent == 0
end

@testset "Elapsed Time Interruption" begin
    top = CoupledDEVS(:top)
    sender_ref = add_submodel!(top, :sender, MockNode; out_ports=(; out=Int), ta=1.5, make_out=true)
    receiver_ref = add_submodel!(top, :receiver, MockNode; inbox=(; in=Int[]), ta=5.0)
    connect!(top, sender_ref.out, receiver_ref.in)

    model = compile_devs(top; logging=julia_test_file)
    simulate_devs!(model, 2.0)
    test_trace_equivalence(julia_test_file, "test/traces/ElapsedTimeInterruption.csv")

    receiver = model[model.path_to_id["top.receiver"]]
    @test receiver.n_external == 1
    @test receiver.last_e == 1.5
end

@testset "Elapsed Time Reset (Multiple Interruptions)" begin
    top = CoupledDEVS(:top)

    s1 = add_submodel!(top, :s1, MockNode; out_ports=(; out=Int), ta=3.0, make_out=true)
    s2 = add_submodel!(top, :s2, MockNode; out_ports=(; out=Int), ta=5.0, make_out=true)

    rec = add_submodel!(top, :rec, MockNode; inbox=(; in=Int[]), ta=10.0)

    connect!(top, s1.out, rec.in)
    connect!(top, s2.out, rec.in)

    model = compile_devs(top; logging=julia_test_file)
    simulate_devs!(model, 4.0)
    test_trace_equivalence(julia_test_file, "test/traces/ElapsedTimeReset1.csv")

    @test model[model.path_to_id["top.rec"]].last_e == 3.0

    simulate_devs!(model, 6.0)
    test_trace_equivalence(julia_test_file, "test/traces/ElapsedTimeReset2.csv")
    @test model[model.path_to_id["top.rec"]].last_e == 2.0
end

@testset "Confluent Transition Test" begin
    top = CoupledDEVS(:top)

    sender_ref = add_submodel!(top, :sender, MockNode;
        out_ports=(; out=Int),
        ta=2.0, make_out=true
    )
    receiver_ref = add_submodel!(top, :receiver, MockNode;
        inbox=(; in=Int[]),
        ta=2.0
    )
    connect!(top, sender_ref.out, receiver_ref.in)

    model = compile_devs(top; logging=julia_test_file)
    simulate_devs!(model)
    test_trace_equivalence(julia_test_file, "test/traces/Confluent.csv")

    sender = model[model.path_to_id["top.sender"]]
    @test sender.n_internal == 1
    @test sender.n_external == 0
    @test sender.n_confluent == 0

    receiver = model[model.path_to_id["top.receiver"]]
    @test receiver.n_internal == 1
    @test receiver.n_external == 1
    @test receiver.n_confluent == 1
end

@testset "Nearly Confluent Transition Test" begin
    top = CoupledDEVS(:top)

    sender_ref = add_submodel!(top, :sender, MockNode;
        out_ports=(; out=Int),
        ta=1.999, make_out=true
    )
    receiver_ref = add_submodel!(top, :receiver, MockNode;
        inbox=(; in=Int[]),
        ta=2.000
    )
    connect!(top, sender_ref.out, receiver_ref.in)

    model = compile_devs(top; logging=julia_test_file)
    simulate_devs!(model)
    test_trace_equivalence(julia_test_file, "test/traces/NearlyConfluent.csv")

    sender = model[model.path_to_id["top.sender"]]
    @test sender.n_internal == 1
    @test sender.n_external == 0
    @test sender.n_confluent == 0

    receiver = model[model.path_to_id["top.receiver"]]
    @test receiver.n_internal == 1
    @test receiver.n_external == 1
    @test receiver.n_confluent == 0
end

@testset "Confluent Output Test" begin
    top = CoupledDEVS(:top)

    sender1_ref = add_submodel!(top, :sender1, MockNode;
        out_ports=(; out=Int),
        ta=2.0, make_out=true
    )
    sender2_ref = add_submodel!(top, :sender2, MockNode;
        inbox=(; in=Int[]),
        out_ports=(; out=Int),
        ta=2.0, make_out=true
    )
    receiver_ref = add_submodel!(top, :receiver, MockNode;
        inbox=(; in=Int[]),
    )
    connect!(top, sender1_ref.out, sender2_ref.in)
    connect!(top, sender2_ref.out, receiver_ref.in)

    model = compile_devs(top; logging=julia_test_file)
    simulate_devs!(model)
    test_trace_equivalence(julia_test_file, "test/traces/ConfluentOutput.csv")

    sender1 = model[model.path_to_id["top.sender1"]]
    @test sender1.n_internal == 1
    @test sender1.n_external == 0
    @test sender1.n_confluent == 0
    @test sender1.n_outputs == 1

    sender2 = model[model.path_to_id["top.sender2"]]
    @test sender2.n_internal == 1
    @test sender2.n_external == 1
    @test sender2.n_confluent == 1
    @test sender2.n_outputs == 1

    receiver = model[model.path_to_id["top.receiver"]]
    @test receiver.n_internal == 0
    @test receiver.n_external == 1
    @test receiver.n_confluent == 0
end