"""
Test shell block
Documentation: CHANGE_X
"""
import pytest
from tests_rpl.test_types.test_map_type import EMPTY_ACTOR_VARIABLES
from tests_rpl.rpltests.rpl_cases import RplCaseProgram, check_success_cases
from tests_rpl.rpltests.expressions import expression_testing
from py_reactor_utils.rpl_building import is_wasm_target
from py_reactor_utils.testing_utils import RETURN_CODE_MINIDUMP
from tests_rpl.rpltests.rpl_testing import (check_compile_issues, check_compile_success, check_run_result,
generate_assert, generate_final_assert, IssueCode, merge_text,
RplTestProgram, FATAL_ERROR, check_program_panic)
class Util:
prohibited_statements: list[str] = [
"send MyActor.Msg >> myactor;",
"wait MyActor.Msg << myactor;",
"discard create MyActor();",
"terminate;",
"send MyActor.Msg >> self;",
]
class TestShell:
prohibited_statements = [
"send MyActor.Msg >> myactor;",
"wait MyActor.Msg << myactor;",
"discard create MyActor();",
"terminate;",
]
class TestNestedBlocks:
"""Validates explicit and implicit shell block nesting."""
success_case_programs = []
def collect_cases_a_1_explicit_nesting(self):
"""Verifies explicit nesting in non-void and void shell blocks."""
# Requirement: RPL-SHELL-NB1
for t, v in expression_testing.ALL_TYPES_VALUES_EXAMPLES.items():
# shared value needs to be unpacked
extractor = '*' if t in expression_testing.SHARED_TYPES else ''
# BUG: RND-3065
# remove condition after bug fix
if t == "enum {Red}":
continue
# case a: non-void shell
self.success_case_programs.append(
RplCaseProgram(
additional_sources=["actor EmptyActor;"],
test_actor_imports="import EmptyActor;",
defs_block=merge_text(
expression_testing.ENUM_TYPE_ALIAS,
expression_testing.EMPTY_ACTOR_VARIABLES,
),
main_body=merge_text(
f"""\
var result = wait shell: {t} {{
var res = wait shell: {t} {{
yield {v};
}};
yield res;
}};
""",
generate_assert(f"{extractor}result == {extractor}{v}", "wrong comparison result"),
"Testing.complete();"
)
)
)
# case b: void shell
self.success_case_programs.append(
RplCaseProgram(
additional_sources=["actor EmptyActor;"],
test_actor_imports="import EmptyActor;",
defs_block=merge_text(
expression_testing.ENUM_TYPE_ALIAS,
expression_testing.EMPTY_ACTOR_VARIABLES,
),
main_body=merge_text(
f"""\
wait shell {{
var res = wait shell: {t} {{
yield {v};
}};
assert {extractor}res == {extractor}{v}, "wrong comparison result";
yield;
}};
""",
"Testing.complete();"
)
)
)
def collect_cases_a_2_implicit_nesting(self):
"""Verifies implicit nesting in non-void and void shell blocks.
Cases for actor types collected manually.
And for others via a python loop.
"""
# Requirement: RPL-SHELL-NB1
for t, v in expression_testing.ALL_TYPES_VALUES_EXAMPLES.items():
# shared value needs to be unpacked
extractor = '*' if t in expression_testing.SHARED_TYPES else ''
# BUG: RND-3065
# remove condition after bug fix
if t == "enum {Red}":
continue
if t in EMPTY_ACTOR_VARIABLES:
# actor types will be added manually after for loop
continue
# case a: non-void shell
self.success_case_programs.append(
RplCaseProgram(
additional_sources=["actor EmptyActor;"],
test_actor_imports="import EmptyActor;",
defs_block=merge_text(
expression_testing.ENUM_TYPE_ALIAS,
expression_testing.EMPTY_ACTOR_VARIABLES,
),
function_defs=[f"""\
def execute_inner_shell(): {t} {{
return wait shell: {t} {{
yield {v};
}};
}}
"""],
main_body=merge_text(
f"""\
var result = wait shell: {t} {{
var res = execute_inner_shell();
yield res;
}};
""",
generate_assert(f"{extractor}result == {extractor}{v}", "wrong comparison result"),
"Testing.complete();"
)
)
)
# case b: void shell
self.success_case_programs.append(
RplCaseProgram(
additional_sources=["actor EmptyActor;"],
test_actor_imports="import EmptyActor;",
defs_block=merge_text(
expression_testing.ENUM_TYPE_ALIAS,
expression_testing.EMPTY_ACTOR_VARIABLES,
),
function_defs=[f"""\
def execute_inner_shell(): {t} [[shell]] {{
return wait shell: {t} {{
yield {v};
}};
}}
"""],
main_body=merge_text(
f"""\
wait shell {{
var res = execute_inner_shell();
assert {extractor}res == {extractor}{v}, "wrong comparison result";
yield;
}};
""",
"Testing.complete();"
)
)
)
# to prevent static storage access error
# add cases with actor types manually
self.success_case_programs.append(
RplCaseProgram(
additional_sources=[
"actor EmptyActor;",
"""\
actor TestActor;
import EmptyActor;
type EnumRed = enum {Red};
def execute_inner_shell_1(actor_var: EmptyActor): EmptyActor {
return wait shell: EmptyActor {
yield actor_var;
};
}
def execute_inner_shell_2(actor_var: actor): actor {
return wait shell: actor {
yield actor_var;
};
}
on create (sender, data: {}) {
var emptyActor = create EmptyActor();
var anyActor: actor = emptyActor as actor;
// case a: non-void shell
var result_1 = wait shell: EmptyActor {
var res = execute_inner_shell_1(emptyActor);
yield res;
};
assert result_1 == emptyActor, "wrong comparison result";
var result_2 = wait shell: actor {
var res = execute_inner_shell_2(anyActor);
yield res;
};
assert result_2 == anyActor, "wrong comparison result";
// case b: void shell
wait shell {
var res = execute_inner_shell_1(emptyActor);
assert res == emptyActor, "wrong comparison result";
yield;
};
wait shell {
var res = execute_inner_shell_2(anyActor);
assert res == anyActor, "wrong comparison result";
yield;
};
}
"""
],
test_actor_imports="import TestActor;",
main_body=merge_text(
"""\
discard create TestActor({});
""",
"Testing.complete();"
)
)
)
def test_success_cases(self):
self.collect_cases_a_1_explicit_nesting()
self.collect_cases_a_2_implicit_nesting()
check_success_cases({f"case_{num}": case for num, case in enumerate(self.success_case_programs, start=1)})
class TestIsolationLayer:
"""Validates using prohibited statements leads to CTE or panics."""
@staticmethod
def test_a_1_prohibited_statements():
# Requirement: RPL-SHELL-IL1
"""Verifies compiler throws E_WrongScope on prohibited statements usage."""
prohibited_statements = Util.prohibited_statements + ["return;"]
for statement_ in prohibited_statements:
program = RplTestProgram(
additional_sources=["""\
actor MyActor;
message Msg contains {};
on Msg(sender, data) {}
""",],
test_actor_imports="""\
import MyActor;
on Msg(sender, data) {}
""",
test_actor_contents="""\
var myactor = create MyActor();
""",
main_body=f"""\
wait shell {{
{statement_}
}};
"""
)
check_compile_issues(program, IssueCode.E_WrongScope)
@staticmethod
def test_b_1_prohibited_statements_indirectly():
# Requirement: RPL-SHELL-IL2
"""Verifies runtime panics on prohibited statements indirect usage.
RplTestProgram creates singleton actor. Therefore, in order to get
complete test scenario we should use an additional .rpl source file and
create a testing actor instance manually.
"""
for statement_ in Util.prohibited_statements:
program = RplTestProgram(
additional_sources=[
"""\
actor MyActor;
message Msg contains {};
on Msg(sender, data) {}
""",
f"""
actor TestActor;
import MyActor;
on Msg(sender, data) {{}}
def use_prohibited_statement(myactor: MyActor) {{
{statement_}
}}
on create(sender, data: {{}}) {{
// to prevent static storage access error in shell
// we should pass the actor instance as an argument
var myactor = create MyActor();
wait shell {{
use_prohibited_statement(myactor);
}};
}}
"""
],
test_actor_imports="import TestActor;",
main_body="discard create TestActor({});"
)
check_compile_success(program)
check_run_result(program,
expected_stderr_substrings=[FATAL_ERROR],
expected_exit_code=RETURN_CODE_MINIDUMP,
expected_stdout='')
@staticmethod
def test_b_2_non_trivial_example():
# Requirement: RPL-SHELL-IL2
"""Verifies runtime successes or panics: non-trivial case.
RplTestProgram creates singleton actor. Therefore, in order to get
complete test scenario we should use an additional .rpl source file and
create a testing actor instance manually.
"""
program_successes = RplTestProgram(
additional_sources=[
"""\
actor MyActor;
""",
"""
actor TestActor;
import MyActor;
import Testing;
def use_prohibited_statement_conditionally(condition: bool): int {
if (condition)
var new_actor = create MyActor();
return 0;
}
on create(sender, data: {}) {
var myactor = create MyActor();
var result = wait shell: int {
yield use_prohibited_statement_conditionally(false);
};
assert result == 0, "wrong shell yield result";
Testing.complete();
}
"""
],
test_actor_imports="import TestActor;",
main_body="discard create TestActor({});"
)
check_compile_success(program_successes)
check_run_result(program_successes)
program_panics = RplTestProgram(
additional_sources=[
"""\
actor MyActor;
""",
"""
actor TestActor;
import MyActor;
import Testing;
def use_prohibited_statement_conditionally(condition: bool): int {
if (condition)
var new_actor = create MyActor();
return 0;
}
on create(sender, data: {}) {
var myactor = create MyActor();
var result = wait shell: int {
yield use_prohibited_statement_conditionally(true);
};
assert result == 0, "wrong shell yield result";
Testing.complete();
}
"""
],
test_actor_imports="import TestActor;",
main_body="discard create TestActor({});"
)
check_compile_success(program_panics)
check_run_result(program_panics,
expected_stderr_substrings=[FATAL_ERROR],
expected_exit_code=RETURN_CODE_MINIDUMP,
expected_stdout='')
def test_b_1_no_result(self):
program = RplTestProgram(
main_body="""\
wait shell {
var a = 100;
};
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
def test_b_2_no_result_error(self):
program = RplTestProgram(
main_body="""\
wait shell {
yield 100;
};
"""
)
check_compile_issues(program, IssueCode.E_ValueNotExpected)
def test_b_3_with_result(self):
program = RplTestProgram(
main_body="""\
var res = wait shell: int {
yield 100;
};
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
def test_b_3_with_result_error(self):
program = RplTestProgram(
main_body="""\
var res = wait shell: int {
var a = 100;
};
"""
)
check_compile_issues(program, IssueCode.E_NoReturn)
def test_c_1_yield_without_expression(self):
program = RplTestProgram(
main_body="""\
var condition = true; // to avoid compiler raise "unreachable code"
wait shell {
if (condition)
yield;
Testing.fail("Should not reach this point.");
};
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
def test_c_2_yield_with_expression(self):
program = RplTestProgram(
main_body=merge_text(
"""\
const value1 = 100;
const value2 = 200;
var res = wait shell: int {
var condition = true; // to avoid compiler raise "unreachable code"
if (condition)
yield value1;
yield value2;
};
""",
generate_final_assert("res == value1", "wrong result: {res} != {value1}")
)
)
check_compile_success(program)
check_run_result(program)
@pytest.mark.skipif(is_wasm_target(), reason="RND-2756")
def test_d_1_exception_propagation(self):
check_program_panic(
"""
wait shell {
var condition = true; // to avoid compiler raise "unreachable code"
if (condition)
assert false, "panic in the shell";
Testing.fail("Should not reach this point.");
};
""",
expected_stderr_substrings=["panic in the shell"]
)
check_program_panic(
"""
var res = wait shell: int {
var condition = true; // to avoid compiler raise "unreachable code"
if (condition)
assert false, "panic in the shell";
Testing.fail("Should not reach this point.");
yield 0;
};
""",
expected_stderr_substrings=["panic in the shell"]
)
@pytest.mark.skipif(is_wasm_target(), reason="RND-2756")
def test_d_2_exception_propagation_from_nested(self):
check_program_panic(
"""
wait shell {
wait shell {
var condition = true; // to avoid compiler raise "unreachable code"
if (condition)
assert false, "panic in the shell";
};
Testing.fail("Should not reach this point.");
};
""",
expected_stderr_substrings=["panic in the shell"]
)
check_program_panic(
"""
discard wait shell: int {
discard wait shell: int {
var condition = true; // to avoid compiler raise "unreachable code"
if (condition)
assert false, "panic in the shell";
Testing.fail("Should not reach this point.");
yield 0;
};
Testing.fail("Should not reach this point.");
yield 0;
};
""",
expected_stderr_substrings=["panic in the shell"]
)
@pytest.mark.skipif(is_wasm_target(), reason="RND-2756")
def test_d_3_exception_propagation_from_nested_function(self):
check_program_panic(
RplTestProgram(
test_actor_contents="""\
def nested_without_result() {
wait shell {
var condition = true; // to avoid compiler raise "unreachable code"
if (condition)
assert false, "panic in the shell";
};
Testing.fail("Should not reach this point.");
}
""",
main_body="""
wait shell {
nested_without_result();
Testing.fail("Should not reach this point.");
};
"""
),
expected_stderr_substrings=["panic in the shell"]
)
check_program_panic(
RplTestProgram(
test_actor_contents="""\
def nested_with_result(): int {
return wait shell: int {
var condition = true; // to avoid compiler raise "unreachable code"
if (condition)
assert false, "panic in the shell";
Testing.fail("Should not reach this point.");
yield 0;
};
}
""",
main_body="""
var res = wait shell: int {
yield nested_with_result();
};
"""
),
expected_stderr_substrings=["panic in the shell"]
)
def test_e_2_nested_captured_argument_yield(self):
program = RplTestProgram(
test_actor_contents="""\
def shelled_argument(arg: int): int {
return wait shell: int {
yield arg;
};
}
""",
main_body=merge_text(
"""\
const argument_value = 300;
var res = wait shell: int {
yield shelled_argument(argument_value);
};
""",
generate_final_assert("res == argument_value", "wrong res {res} != {argument_value}")
)
)
check_compile_success(program)
check_run_result(program)
def test_f_1_shell_attribute_error(self):
program = RplTestProgram(
test_actor_contents="""\
def shell_attributed() [[shell]] {}
""",
main_body="""\
wait shell {
shell_attributed(); // Okay
};
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
def test_f_2_shell_attribute_error(self):
program = RplTestProgram(
test_actor_contents="""\
def shell_attributed() [[shell]] {}
""",
main_body="""\
shell_attributed(); // Calling not from a shell
"""
)
check_compile_issues(program, IssueCode.E_WrongScope)
program = RplTestProgram(
test_actor_contents="""\
def shell_attributed() [[shell]] {}
def func() { // doesn't inherit [[shell]] attribute
shell_attributed(); // Calling not from a shell
}
""",
main_body="""\
func();
"""
)
check_compile_issues(program, IssueCode.E_WrongScope)
def test_f_3_shell_attribute_prohibited_statements(self):
def check_prohibited_statement(statement_: str):
program = RplTestProgram(
additional_sources=["""\
actor MyActor;
message Msg contains {};
on Msg(sender, data) {}
""", f"""\
actor TestActor;
import MyActor;
def shell_attributed(myactor: MyActor) [[shell]] {{
{statement_}
}}
"""],
test_actor_imports="",
main_body=""
)
check_compile_issues(program, IssueCode.E_WrongScope)
for statement in TestShell.prohibited_statements:
check_prohibited_statement(statement)
def test_g_implicit_shell_block(self):
pass # Untestable (yet)
def test_h_1_captured_all(self):
program = RplTestProgram(
test_actor_contents="""\
var global_variable = 100;
def shelled_argument(arg: int): int {
return wait shell: int {
yield arg;
};
}
""",
main_body=merge_text(
"""\
var res1 = wait shell: int {
yield global_variable;
};
var local_variable = 200;
var res2 = wait shell: int {
yield local_variable;
};
const argument_value = 300;
var res3 = shelled_argument(argument_value);
""",
generate_assert("res1 == global_variable", "wrong res1 {res1}"),
generate_assert("res2 == local_variable", "wrong res2 {res2}"),
generate_final_assert("res3 == argument_value", "wrong res3 {res3}")
)
)
check_compile_success(program)
check_run_result(program)
def test_h_1_captured_global_variable_writing_error(self):
program = RplTestProgram(
test_actor_contents="""\
var global_variable = 100;
""",
main_body="""\
var res = wait shell: int {
global_variable = global_variable + 1;
yield global_variable;
};
Testing.complete();
"""
)
check_compile_issues(program, IssueCode.E_ReadOnly)
def test_h_1_captured_local_variable_writing_error(self):
program = RplTestProgram(
main_body="""\
var local_variable = 200;
var res = wait shell: int {
local_variable = local_variable + 1;
yield local_variable;
};
Testing.complete();
"""
)
check_compile_issues(program, IssueCode.E_ReadOnly)
def test_h_1_captured_argument_writing_error(self):
program = RplTestProgram(
test_actor_contents="""\
def func(arg: int): int {
return wait shell: int {
arg = arg + 1;
yield arg;
};
}
""",
main_body="""\
func(100);
Testing.complete();
"""
)
check_compile_issues(program, IssueCode.E_ReadOnly)
def test_h_2_access_actor_state_indirectly_runtime(self):
program = RplTestProgram(
params=dict(test_function_name='str'),
test_actor_contents="""\
var global_x = 0;
def read_access(): int {
return global_x;
}
def write_access() {
global_x = 1;
}
""",
main_body="""\
if (test_function_name == "read_access") {
wait shell {
discard read_access();
};
}
if (test_function_name == "write_access") {
wait shell {
write_access();
};
}
Testing.fail("Should not reach this point.");
"""
)
check_compile_success(program)
check_run_result(program, args=dict(test_function_name='read_access'),
expected_stderr_substrings=[FATAL_ERROR],
expected_exit_code=RETURN_CODE_MINIDUMP,
expected_stdout='')
check_run_result(program, args=dict(test_function_name='write_access'),
expected_stderr_substrings=[FATAL_ERROR],
expected_exit_code=RETURN_CODE_MINIDUMP,
expected_stdout='')
# Don't test Execution
# Don't test ACID
# Don't test Crash Recovery
class TestScenarios:
def test_a_scenario_1(self):
pass # Don't test scenario1
db_api = """\
type DB = {}; // implementation-specific
def connect(locator: str) : DB [[shell]] {
return {};
}
def execute(db: DB, query: str) [[shell]] {
}
"""
def test_b_scenario_2(self):
f_function_code = """\
def f() : DB [[shell]] {
return connect("mysql");
}
"""
program = RplTestProgram(
test_actor_contents=merge_text(
TestScenarios.db_api,
f_function_code
),
main_body="""\
// Option 1
f(); // 1
"""
)
check_compile_issues(program, IssueCode.E_WrongScope)
program = RplTestProgram(
test_actor_contents=merge_text(
TestScenarios.db_api,
f_function_code
),
main_body="""\
// Option 2
wait shell {
var db = f(); // 2
db.execute("delete * from test");
};
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
def test_c_scenario_3(self):
actor_source = """\
actor Actor;
"""
program = RplTestProgram(
additional_sources=[actor_source],
test_actor_imports="import Actor;",
test_actor_contents="""\
def random(): int {
return 1;
}
def f() {
var x = random();
var a = create Actor(); // 1
}
""",
main_body="""\
wait shell {
f();
};
Testing.fail("Should not reach this point.");
"""
)
check_compile_success(program)
check_run_result(program,
expected_stderr_substrings=[FATAL_ERROR],
expected_exit_code=RETURN_CODE_MINIDUMP,
expected_stdout='')
program = RplTestProgram(
additional_sources=[actor_source],
test_actor_imports="import Actor;",
test_actor_contents=merge_text(
TestScenarios.db_api,
"""\
def g() [[shell]] {
var x = connect("mysql");
var a = create Actor(); // 2
}
"""
),
main_body="""\
wait shell {
g();
};
Testing.complete();
"""
)
check_compile_issues(program, IssueCode.E_WrongScope)
def test_d_scenario_4(self):
program = RplTestProgram(
params=dict(direct='bool'),
test_actor_contents="""\
var global_x: int = 0; // global state variable
def f() {
var x = global_x; // 1
wait shell { var x = global_x; }; // 2
}
""",
main_body="""\
if (direct) {
// Case 1
f(); // 3
} else {
// Case 2
wait shell { f(); }; // 4
}
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program, args=dict(direct='true'))
check_run_result(program, args=dict(direct='false'),
expected_stderr_substrings=[FATAL_ERROR],
expected_exit_code=RETURN_CODE_MINIDUMP,
expected_stdout='')
# FUTURE: shell rvalue expression tests
class TestExtra:
def test_a_1_seq_shells(self):
program = RplTestProgram(
main_body=merge_text(
"""\
wait shell {
var a = 1;
var b = a + 2;
yield;
};
wait shell {
var a = 1+1;
};
var a = wait shell: int {
var n = 10;
yield n;
};
var b = wait shell: str {
var s = "hello";
yield s;
};
""",
generate_final_assert("a == 10 and b == \"hello\"", "wrong result")
)
)
check_compile_success(program)
check_run_result(program)
def test_a_2_seq_native(self):
program = RplTestProgram(
main_body="""\
discard Testing.generate_random_int();
wait shell { discard Testing.multiply_by_2(1); };
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
def test_b_1_seq_same_function_shell(self):
program = RplTestProgram(
test_actor_contents="""\
def f() {
wait shell {
var a = 10;
var b = 20;
var c = a + b;
};
}
""",
main_body="""\
f();
f();
Testing.complete();
""",
)
check_compile_success(program)
check_run_result(program)
def test_b_2_seq_same_native_function_shell(self):
program = RplTestProgram(
main_body="""\
discard Testing.generate_random_int();
discard Testing.generate_random_int();
wait shell {
discard Testing.multiply_by_2(1);
discard Testing.multiply_by_2(1);
};
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
def test_c_all_coro_shell(self):
program = RplTestProgram(
test_actor_contents="""\
def f() {
wait shell {
Testing.empty_coroutine();
};
}
""",
main_body="""\
wait shell {
Testing.empty_coroutine(); // 1
};
f(); // 2
wait shell {
f(); // 3
};
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
def test_d_1_all_native_call(self):
program = RplTestProgram(
main_body=merge_text("""\
wait shell {
var a = Testing.generate_random_int();
var b = a + 10;
};
var a = wait shell: int {
yield Testing.generate_random_int();
};
Testing.complete();
""")
)
check_compile_success(program)
check_run_result(program)
@pytest.mark.skip(reason="need native function throwing exception")
def test_d_2_exception_native_call(self):
pass
@pytest.mark.skip(reason="need native function with shell that calls another native function with shell")
def test_d_3_1_nested_shell_native_call(self):
pass
def test_d_3_2_native_call_from_explicit_shell(self):
program = RplTestProgram(
main_body="""\
var random = Testing.generate_random_int();
var r = wait shell: int {
yield Testing.generate_random_int();
};
Testing.complete();
"""
)
check_compile_success(program)
check_run_result(program)
@pytest.mark.skip(reason="TODO")
def test_e(self):
pass
