diff --git a/tests/test_compile_armv7.py b/tests/test_compile_armv7.py index f36a76c..4d2ee4d 100644 --- a/tests/test_compile_armv7.py +++ b/tests/test_compile_armv7.py @@ -1,32 +1,45 @@ -from copapy import NumLike -from copapy.backend import Store, compile_to_dag, add_read_value_remote -import subprocess -from copapy import _binwrite -import copapy.backend as backend -import copapy as cp import os +import subprocess import warnings + import pytest -if os.name == 'nt': +import copapy as cp +import copapy.backend as backend +from copapy import NumLike, _binwrite +from copapy.backend import Store, add_read_value_remote, compile_to_dag + +if os.name == "nt": # On Windows wsl and qemu-user is required: # sudo apt install qemu-user - qemu_command = ['wsl', 'qemu-arm'] + qemu_command = ["wsl", "qemu-arm"] else: - qemu_command = ['qemu-arm'] + qemu_command = ["qemu-arm"] def run_command(command: list[str]) -> str: - result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf8', check=False) - assert result.returncode != 11, f"SIGSEGV (segmentation fault)\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" - assert result.returncode == 0, f"\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + result = subprocess.run( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + encoding="utf8", + check=False, + ) + assert result.returncode != 11, ( + f"SIGSEGV (segmentation fault)\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + ) + assert result.returncode == 0, ( + f"\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + ) return result.stdout def check_for_qemu() -> bool: - command = qemu_command + ['--version'] + command = qemu_command + ["--version"] try: - result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) + result = subprocess.run( + command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False + ) except Exception: return False return result.returncode == 0 @@ -54,12 +67,12 @@ def test_compile(): out = [Store(r) for r in ret] - sdb = backend.stencil_db_from_package('armv7') + sdb = backend.stencil_db_from_package("armv7") il, variables = compile_to_dag(out, sdb) # run program command il.write_com(_binwrite.Command.RUN_PROG) - #il.write_com(_binwrite.Command.DUMP_CODE) + # il.write_com(_binwrite.Command.DUMP_CODE) for v in ret: assert isinstance(v, cp.value) @@ -67,23 +80,29 @@ def test_compile(): il.write_com(_binwrite.Command.END_COM) - #print('* Data to runner:') - #il.print() + # print('* Data to runner:') + # il.print() - il.to_file('build/runner/test-armv7.copapy') + il.to_file("build/runner/test-armv7.copapy") if not check_for_qemu(): - warnings.warn("qemu-armv7 not found, aarch64 test skipped!", UserWarning) - elif not os.path.isfile('build/runner/coparun-armv7'): + warnings.warn("qemu-armv7 not found, test skipped!", UserWarning) + elif "wsl" in qemu_command: + warnings.warn("qemu-armv7 seams not work on wsl1, test skipped!", UserWarning) + elif not os.path.isfile("build/runner/coparun-armv7"): warnings.warn("armv7 runner not found, aarch64 test skipped!", UserWarning) else: - command = ['build/runner/coparun-armv7', 'build/runner/test-armv7.copapy', 'build/runner/test-armv7.copapy.bin'] + command = [ + "build/runner/coparun-armv7", + "build/runner/test-armv7.copapy", + "build/runner/test-armv7.copapy.bin", + ] result = run_command(qemu_command + command) - print('* Output from runner:\n--') + print("* Output from runner:\n--") print(result) - print('--') + print("--") - assert 'Return value: 1' in result + assert "Return value: 1" in result # Compare to x86_64 reference results assert " size=4 data=24 00 00 00" in result @@ -92,5 +111,5 @@ def test_compile(): if __name__ == "__main__": - #test_example() + # test_example() test_compile() diff --git a/tests/test_ops_armv6.py b/tests/test_ops_armv6.py index 75252fb..4be1cfe 100644 --- a/tests/test_ops_armv6.py +++ b/tests/test_ops_armv6.py @@ -1,21 +1,22 @@ -from copapy import NumLike, iif, value -from copapy.backend import Store, compile_to_dag, add_read_value_remote -import subprocess -from copapy import _binwrite -import copapy.backend as backend import os -import warnings import re import struct -import pytest -import copapy as cp +import subprocess +import warnings -if os.name == 'nt': +import pytest + +import copapy as cp +import copapy.backend as backend +from copapy import NumLike, _binwrite, iif, value +from copapy.backend import Store, add_read_value_remote, compile_to_dag + +if os.name == "nt": # On Windows wsl and qemu-user is required: # sudo apt install qemu-user - qemu_command = ['wsl', 'qemu-arm'] + qemu_command = ["wsl", "qemu-arm"] else: - qemu_command = ['qemu-arm'] + qemu_command = ["qemu-arm"] def parse_results(log_text: str) -> dict[int, bytes]: @@ -24,8 +25,8 @@ def parse_results(log_text: str) -> dict[int, bytes]: var_dict: dict[int, bytes] = {} for match in matches: - value_str: list[str] = match.group(3).strip().split(' ') - #print('--', value_str) + value_str: list[str] = match.group(3).strip().split(" ") + # print('--', value_str) value = bytes(int(v, base=16) for v in value_str) if len(value) <= 8: var_dict[int(match.group(1))] = value @@ -34,26 +35,49 @@ def parse_results(log_text: str) -> dict[int, bytes]: def run_command(command: list[str]) -> str: - result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf8', check=False) - assert result.returncode != 11, f"SIGSEGV (segmentation fault)\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" - assert result.returncode == 0, f"\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + result = subprocess.run( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + encoding="utf8", + check=False, + ) + assert result.returncode != 11, ( + f"SIGSEGV (segmentation fault)\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + ) + assert result.returncode == 0, ( + f"\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + ) return result.stdout def check_for_qemu() -> bool: - command = qemu_command + ['--version'] + command = qemu_command + ["--version"] try: - result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) + result = subprocess.run( + command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False + ) except Exception: return False return result.returncode == 0 def function1(c1: NumLike) -> list[NumLike]: - return [c1 / 4, c1 / -4, c1 // 4, c1 // -4, (c1 * -1) // 4, - c1 * 4, c1 * -4, - c1 + 4, c1 - 4, - c1 > 2, c1 > 100, c1 < 4, c1 < 100] + return [ + c1 / 4, + c1 / -4, + c1 // 4, + c1 // -4, + (c1 * -1) // 4, + c1 * 4, + c1 * -4, + c1 + 4, + c1 - 4, + c1 > 2, + c1 > 100, + c1 < 4, + c1 < 100, + ] def function2(c1: NumLike) -> list[NumLike]: @@ -77,11 +101,13 @@ def function6(c1: NumLike) -> list[NumLike]: def iiftests(c1: NumLike) -> list[NumLike]: - return [iif(c1 > 5, 8, 9), - iif(c1 < 5, 8.5, 9.5), - iif(1 > 5, 3.3, 8.8) + c1, - iif(1 < 5, c1 * 3.3, 8.8), - iif(c1 < 5, c1 * 3.3, 8.8)] + return [ + iif(c1 > 5, 8, 9), + iif(c1 < 5, 8.5, 9.5), + iif(1 > 5, 3.3, 8.8) + c1, + iif(1 < 5, c1 * 3.3, 8.8), + iif(c1 < 5, c1 * 3.3, 8.8), + ] @pytest.mark.runner @@ -90,59 +116,90 @@ def test_compile(): c_f = value(1.111) c_b = value(True) - ret_test = function1(c_i) + function1(c_f) + function2(c_i) + function2(c_f) + function3(c_i) + function4(c_i) + function5(c_b) + [value(9) % 2] + iiftests(c_i) + iiftests(c_f) + [cp.asin(c_i/10)] - ret_ref = function1(9) + function1(1.111) + function2(9) + function2(1.111) + function3(9) + function4(9) + function5(True) + [9 % 2] + iiftests(9) + iiftests(1.111) + [cp.asin(9/10)] + ret_test = ( + function1(c_i) + + function1(c_f) + + function2(c_i) + + function2(c_f) + + function3(c_i) + + function4(c_i) + + function5(c_b) + + [value(9) % 2] + + iiftests(c_i) + + iiftests(c_f) + + [cp.asin(c_i / 10)] + ) + ret_ref = ( + function1(9) + + function1(1.111) + + function2(9) + + function2(1.111) + + function3(9) + + function4(9) + + function5(True) + + [9 % 2] + + iiftests(9) + + iiftests(1.111) + + [cp.asin(9 / 10)] + ) - #ret_test = (c_i * 100 // 5, c_f * 10 // 5) - #ret_ref = (9 * 100 // 5, 1.111 * 10 // 5) + # ret_test = (c_i * 100 // 5, c_f * 10 // 5) + # ret_ref = (9 * 100 // 5, 1.111 * 10 // 5) out = [Store(r) for r in ret_test] - sdb = backend.stencil_db_from_package('armv6') + sdb = backend.stencil_db_from_package("armv6") dw, variables = compile_to_dag(out, sdb) - #dw.write_com(_binwrite.Command.READ_DATA) - #dw.write_int(0) - #dw.write_int(28) + # dw.write_com(_binwrite.Command.READ_DATA) + # dw.write_int(0) + # dw.write_int(28) # run program command dw.write_com(_binwrite.Command.RUN_PROG) - #dw.write_com(_binwrite.Command.DUMP_CODE) + # dw.write_com(_binwrite.Command.DUMP_CODE) for v in ret_test: assert isinstance(v, value) add_read_value_remote(dw, variables, v.net) - #dw.write_com(_binwrite.Command.READ_DATA) - #dw.write_int(0) - #dw.write_int(28) + # dw.write_com(_binwrite.Command.READ_DATA) + # dw.write_int(0) + # dw.write_int(28) dw.write_com(_binwrite.Command.END_COM) - #print('* Data to runner:') - #dw.print() + # print('* Data to runner:') + # dw.print() - dw.to_file('build/runner/test-armv6.copapy') + dw.to_file("build/runner/test-armv6.copapy") if not check_for_qemu(): warnings.warn("qemu-armv6 not found, armv6 test skipped!", UserWarning) return - if not os.path.isfile('build/runner/coparun-armv6'): + if "wsl" in qemu_command: + warnings.warn("qemu-armv6 seams not work on wsl1, test skipped!", UserWarning) + return + if not os.path.isfile("build/runner/coparun-armv6"): warnings.warn("armv6 runner not found, armv6 test skipped!", UserWarning) return - command = qemu_command + ['build/runner/coparun-armv6', 'build/runner/test-armv6.copapy'] + ['build/runner/test-armv6.copapy.bin'] - #try: + command = ( + qemu_command + + ["build/runner/coparun-armv6", "build/runner/test-armv6.copapy"] + + ["build/runner/test-armv6.copapy.bin"] + ) + # try: result = run_command(command) - #except FileNotFoundError: + # except FileNotFoundError: # warnings.warn(f"Test skipped, executable not found.", UserWarning) # return - #print('* Output from runner:\n--') - #print(result) - #print('--') + # print('* Output from runner:\n--') + # print(result) + # print('--') - assert 'Return value: 1' in result + assert "Return value: 1" in result result_data = parse_results(result) @@ -150,22 +207,25 @@ def test_compile(): assert isinstance(test, value) address = variables[test.net][0] data = result_data[address] - if test.dtype == 'int': + if test.dtype == "int": val = int.from_bytes(data, sdb.byteorder, signed=True) - elif test.dtype == 'bool': + elif test.dtype == "bool": val = bool.from_bytes(data, sdb.byteorder) - elif test.dtype == 'float': - en = {'little': '<', 'big': '>'}[sdb.byteorder] - val = struct.unpack(en + 'f', data)[0] + elif test.dtype == "float": + en = {"little": "<", "big": ">"}[sdb.byteorder] + val = struct.unpack(en + "f", data)[0] assert isinstance(val, float) else: raise Exception(f"Unknown type: {test.dtype}") - print('+', val, ref, test.dtype, f" addr={address}") + print("+", val, ref, test.dtype, f" addr={address}") for t in (int, float, bool): - assert isinstance(val, t) == isinstance(ref, t), f"Result type does not match for {val} and {ref}" - assert val == pytest.approx(ref, 1e-5), f"Result does not match: {val} and reference: {ref}" # pyright: ignore[reportUnknownMemberType] + assert isinstance(val, t) == isinstance(ref, t), ( + f"Result type does not match for {val} and {ref}" + ) + assert val == pytest.approx(ref, 1e-5), ( + f"Result does not match: {val} and reference: {ref}" + ) # pyright: ignore[reportUnknownMemberType] if __name__ == "__main__": - #test_compile() - test_slow_31bit_int_list_hash() + test_compile() diff --git a/tests/test_ops_armv7.py b/tests/test_ops_armv7.py index ee0c5b7..ea2b4b0 100644 --- a/tests/test_ops_armv7.py +++ b/tests/test_ops_armv7.py @@ -1,21 +1,22 @@ -from copapy import NumLike, iif, value -from copapy.backend import Store, compile_to_dag, add_read_value_remote -import subprocess -from copapy import _binwrite -import copapy.backend as backend import os -import warnings import re import struct -import pytest -import copapy as cp +import subprocess +import warnings -if os.name == 'nt': +import pytest + +import copapy as cp +import copapy.backend as backend +from copapy import NumLike, _binwrite, iif, value +from copapy.backend import Store, add_read_value_remote, compile_to_dag + +if os.name == "nt": # On Windows wsl and qemu-user is required: # sudo apt install qemu-user - qemu_command = ['wsl', 'qemu-arm'] + qemu_command = ["wsl", "qemu-arm"] else: - qemu_command = ['qemu-arm'] + qemu_command = ["qemu-arm"] def parse_results(log_text: str) -> dict[int, bytes]: @@ -24,8 +25,8 @@ def parse_results(log_text: str) -> dict[int, bytes]: var_dict: dict[int, bytes] = {} for match in matches: - value_str: list[str] = match.group(3).strip().split(' ') - #print('--', value_str) + value_str: list[str] = match.group(3).strip().split(" ") + # print('--', value_str) value = bytes(int(v, base=16) for v in value_str) if len(value) <= 8: var_dict[int(match.group(1))] = value @@ -34,26 +35,49 @@ def parse_results(log_text: str) -> dict[int, bytes]: def run_command(command: list[str]) -> str: - result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf8', check=False) - assert result.returncode != 11, f"SIGSEGV (segmentation fault)\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" - assert result.returncode == 0, f"\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + result = subprocess.run( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + encoding="utf8", + check=False, + ) + assert result.returncode != 11, ( + f"SIGSEGV (segmentation fault)\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + ) + assert result.returncode == 0, ( + f"\n -Error occurred: {result.stderr}\n -Output: {result.stdout}" + ) return result.stdout def check_for_qemu() -> bool: - command = qemu_command + ['--version'] + command = qemu_command + ["--version"] try: - result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) + result = subprocess.run( + command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False + ) except Exception: return False return result.returncode == 0 def function1(c1: NumLike) -> list[NumLike]: - return [c1 / 4, c1 / -4, c1 // 4, c1 // -4, (c1 * -1) // 4, - c1 * 4, c1 * -4, - c1 + 4, c1 - 4, - c1 > 2, c1 > 100, c1 < 4, c1 < 100] + return [ + c1 / 4, + c1 / -4, + c1 // 4, + c1 // -4, + (c1 * -1) // 4, + c1 * 4, + c1 * -4, + c1 + 4, + c1 - 4, + c1 > 2, + c1 > 100, + c1 < 4, + c1 < 100, + ] def function2(c1: NumLike) -> list[NumLike]: @@ -77,11 +101,13 @@ def function6(c1: NumLike) -> list[NumLike]: def iiftests(c1: NumLike) -> list[NumLike]: - return [iif(c1 > 5, 8, 9), - iif(c1 < 5, 8.5, 9.5), - iif(1 > 5, 3.3, 8.8) + c1, - iif(1 < 5, c1 * 3.3, 8.8), - iif(c1 < 5, c1 * 3.3, 8.8)] + return [ + iif(c1 > 5, 8, 9), + iif(c1 < 5, 8.5, 9.5), + iif(1 > 5, 3.3, 8.8) + c1, + iif(1 < 5, c1 * 3.3, 8.8), + iif(c1 < 5, c1 * 3.3, 8.8), + ] @pytest.mark.runner @@ -90,59 +116,90 @@ def test_compile(): c_f = value(1.111) c_b = value(True) - ret_test = function1(c_i) + function1(c_f) + function2(c_i) + function2(c_f) + function3(c_i) + function4(c_i) + function5(c_b) + [value(9) % 2] + iiftests(c_i) + iiftests(c_f) + [cp.asin(c_i/10)] - ret_ref = function1(9) + function1(1.111) + function2(9) + function2(1.111) + function3(9) + function4(9) + function5(True) + [9 % 2] + iiftests(9) + iiftests(1.111) + [cp.asin(9/10)] + ret_test = ( + function1(c_i) + + function1(c_f) + + function2(c_i) + + function2(c_f) + + function3(c_i) + + function4(c_i) + + function5(c_b) + + [value(9) % 2] + + iiftests(c_i) + + iiftests(c_f) + + [cp.asin(c_i / 10)] + ) + ret_ref = ( + function1(9) + + function1(1.111) + + function2(9) + + function2(1.111) + + function3(9) + + function4(9) + + function5(True) + + [9 % 2] + + iiftests(9) + + iiftests(1.111) + + [cp.asin(9 / 10)] + ) - #ret_test = (c_i * 100 // 5, c_f * 10 // 5) - #ret_ref = (9 * 100 // 5, 1.111 * 10 // 5) + # ret_test = (c_i * 100 // 5, c_f * 10 // 5) + # ret_ref = (9 * 100 // 5, 1.111 * 10 // 5) out = [Store(r) for r in ret_test] - sdb = backend.stencil_db_from_package('armv7') + sdb = backend.stencil_db_from_package("armv7") dw, variables = compile_to_dag(out, sdb) - #dw.write_com(_binwrite.Command.READ_DATA) - #dw.write_int(0) - #dw.write_int(28) + # dw.write_com(_binwrite.Command.READ_DATA) + # dw.write_int(0) + # dw.write_int(28) # run program command dw.write_com(_binwrite.Command.RUN_PROG) - #dw.write_com(_binwrite.Command.DUMP_CODE) + # dw.write_com(_binwrite.Command.DUMP_CODE) for v in ret_test: assert isinstance(v, value) add_read_value_remote(dw, variables, v.net) - #dw.write_com(_binwrite.Command.READ_DATA) - #dw.write_int(0) - #dw.write_int(28) + # dw.write_com(_binwrite.Command.READ_DATA) + # dw.write_int(0) + # dw.write_int(28) dw.write_com(_binwrite.Command.END_COM) - #print('* Data to runner:') - #dw.print() + # print('* Data to runner:') + # dw.print() - dw.to_file('build/runner/test-armv7.copapy') + dw.to_file("build/runner/test-armv7.copapy") if not check_for_qemu(): warnings.warn("qemu-armv7 not found, armv7 test skipped!", UserWarning) return - if not os.path.isfile('build/runner/coparun-armv7'): + if "wsl" in qemu_command: + warnings.warn("qemu-armv7 seams not work on wsl1, test skipped!", UserWarning) + return + if not os.path.isfile("build/runner/coparun-armv7"): warnings.warn("armv7 runner not found, armv7 test skipped!", UserWarning) return - command = qemu_command + ['build/runner/coparun-armv7', 'build/runner/test-armv7.copapy'] + ['build/runner/test-armv7.copapy.bin'] - #try: + command = ( + qemu_command + + ["build/runner/coparun-armv7", "build/runner/test-armv7.copapy"] + + ["build/runner/test-armv7.copapy.bin"] + ) + # try: result = run_command(command) - #except FileNotFoundError: + # except FileNotFoundError: # warnings.warn(f"Test skipped, executable not found.", UserWarning) # return - print('* Output from runner:\n--') + print("* Output from runner:\n--") print(result) - print('--') + print("--") - assert 'Return value: 1' in result + assert "Return value: 1" in result result_data = parse_results(result) @@ -150,22 +207,25 @@ def test_compile(): assert isinstance(test, value) address = variables[test.net][0] data = result_data[address] - if test.dtype == 'int': + if test.dtype == "int": val = int.from_bytes(data, sdb.byteorder, signed=True) - elif test.dtype == 'bool': + elif test.dtype == "bool": val = bool.from_bytes(data, sdb.byteorder) - elif test.dtype == 'float': - en = {'little': '<', 'big': '>'}[sdb.byteorder] - val = struct.unpack(en + 'f', data)[0] + elif test.dtype == "float": + en = {"little": "<", "big": ">"}[sdb.byteorder] + val = struct.unpack(en + "f", data)[0] assert isinstance(val, float) else: raise Exception(f"Unknown type: {test.dtype}") - print('+', val, ref, test.dtype, f" addr={address}") + print("+", val, ref, test.dtype, f" addr={address}") for t in (int, float, bool): - assert isinstance(val, t) == isinstance(ref, t), f"Result type does not match for {val} and {ref}" - assert val == pytest.approx(ref, 1e-5), f"Result does not match: {val} and reference: {ref}" # pyright: ignore[reportUnknownMemberType] + assert isinstance(val, t) == isinstance(ref, t), ( + f"Result type does not match for {val} and {ref}" + ) + assert val == pytest.approx(ref, 1e-5), ( + f"Result does not match: {val} and reference: {ref}" + ) # pyright: ignore[reportUnknownMemberType] if __name__ == "__main__": - #test_example() - test_slow_31bit_int_list_hash() + test_compile()