diff --git a/tests/helpers/local_check.py b/tests/helpers/local_check.py new file mode 100644 index 0000000..b704b33 --- /dev/null +++ b/tests/helpers/local_check.py @@ -0,0 +1,166 @@ +# coding: utf-8 +from .os_ops_helpers import OsOpsHelpers +from .os_ops_helpers import OsOperations + +import os + + +class LocalCheck: + @staticmethod + def check_path_exists( + os_ops: OsOperations, + path: str, + ) -> None: + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + if not OsOpsHelpers.is_localhost(os_ops): + return + + if os.path.exists(path): + return + + err_msg = "[LocalCheck] Local path [{}] does not exist.".format( + path, + ) + raise RuntimeError(err_msg) + + # -------------------------------------------------------------------- + @staticmethod + def check_path_does_not_exists( + os_ops: OsOperations, + path: str, + ) -> None: + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + if not OsOpsHelpers.is_localhost(os_ops): + return + + if not os.path.exists(path): + return + + err_msg = "[LocalCheck] Local path [{}] exists.".format( + path, + ) + raise RuntimeError(err_msg) + + # -------------------------------------------------------------------- + @staticmethod + def check_isdir( + os_ops: OsOperations, + path: str, + ) -> None: + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + if not OsOpsHelpers.is_localhost(os_ops): + return + + if os.path.isdir(path): + return + + err_msg = "[LocalCheck] Local path [{}] is not dir.".format( + path, + ) + raise RuntimeError(err_msg) + + # -------------------------------------------------------------------- + @staticmethod + def check_not_isdir( + os_ops: OsOperations, + path: str, + ) -> None: + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + if not OsOpsHelpers.is_localhost(os_ops): + return + + if not os.path.isdir(path): + return + + err_msg = "[LocalCheck] Local path [{}] is dir.".format( + path, + ) + raise RuntimeError(err_msg) + + # -------------------------------------------------------------------- + @staticmethod + def check_isfile( + os_ops: OsOperations, + path: str, + ) -> None: + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + if not OsOpsHelpers.is_localhost(os_ops): + return + + if os.path.isfile(path): + return + + err_msg = "[LocalCheck] Local path [{}] is not file.".format( + path, + ) + raise RuntimeError(err_msg) + + # -------------------------------------------------------------------- + @staticmethod + def check_not_isfile( + os_ops: OsOperations, + path: str, + ) -> None: + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + if not OsOpsHelpers.is_localhost(os_ops): + return + + if not os.path.isfile(path): + return + + err_msg = "[LocalCheck] Local path [{}] is file.".format( + path, + ) + raise RuntimeError(err_msg) + + # -------------------------------------------------------------------- + @staticmethod + def check_path_is_abs( + os_ops: OsOperations, + path: str, + ) -> None: + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + if not OsOpsHelpers.is_localhost(os_ops): + return + + if os.path.isabs(path): + return + + err_msg = "[LocalCheck] Local path [{}] is not abs.".format( + path, + ) + raise RuntimeError(err_msg) + + # -------------------------------------------------------------------- + @staticmethod + def check_path_is_not_abs( + os_ops: OsOperations, + path: str, + ) -> None: + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + if not OsOpsHelpers.is_localhost(os_ops): + return + + if not os.path.isabs(path): + return + + err_msg = "[LocalCheck] Local path [{}] is abs.".format( + path, + ) + raise RuntimeError(err_msg) diff --git a/tests/helpers/os_ops_helpers.py b/tests/helpers/os_ops_helpers.py new file mode 100644 index 0000000..1c27c48 --- /dev/null +++ b/tests/helpers/os_ops_helpers.py @@ -0,0 +1,18 @@ +# coding: utf-8 +from src.os_ops import OsOperations + + +class OsOpsHelpers: + @staticmethod + def is_localhost(os_ops: OsOperations) -> bool: + assert isinstance(os_ops, OsOperations) + + host = os_ops.host + + if host == "127.0.0.1": + return True + + if host == "localhost": + return True + + return False diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 520b405..764f3a2 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -5,16 +5,15 @@ from tests.helpers.global_data import OsOpsDescrs from tests.helpers.global_data import OsOperations from tests.helpers.run_conditions import RunConditions +from tests.helpers.local_check import LocalCheck +from tests.helpers.local_check import OsOpsHelpers import os import sys import pytest import re -import tempfile import logging -import socket -import threading import typing import uuid import subprocess @@ -22,6 +21,7 @@ import time import signal as os_signal import dataclasses +import random from src.exceptions import InvalidOperationException from src.exceptions import ExecUtilException @@ -433,21 +433,16 @@ def test_makedirs_and_rmdirs_success( RunConditions.skip_if_windows() - cmd = "pwd" - stdout = os_ops.exec_command(cmd, encoding='utf-8') - assert type(stdout) is str - pwd = stdout.strip() - - path = "{}/test_dir".format(pwd) + path = "/tmp/testgres-os_ops-test_dir-{}".format(uuid.uuid4().bytes.hex()) # Test makedirs os_ops.makedirs(path) - assert os.path.exists(path) + LocalCheck.check_path_exists(os_ops, path) assert os_ops.path_exists(path) # Test rmdirs os_ops.rmdirs(path) - assert not os.path.exists(path) + LocalCheck.check_path_does_not_exists(os_ops, path) assert not os_ops.path_exists(path) return @@ -537,7 +532,10 @@ def test_path_exists_true__file( RunConditions.skip_if_windows() - assert os_ops.path_exists(__file__) is True + filename = "/bin/sh" + + LocalCheck.check_path_exists(os_ops, filename) + assert os_ops.path_exists(filename) is True return def test_path_exists_false__directory( @@ -588,9 +586,12 @@ def test_mkdtemp__default( path = os_ops.mkdtemp() logging.info("Path is [{0}].".format(path)) - assert os.path.exists(path) - os.rmdir(path) - assert not os.path.exists(path) + LocalCheck.check_path_exists(os_ops, path) + assert os_ops.path_exists(path) + assert os_ops.isdir(path) + os_ops.rmdir(path) + LocalCheck.check_path_does_not_exists(os_ops, path) + assert not os_ops.path_exists(path) return def test_mkdtemp__custom( @@ -606,10 +607,13 @@ def test_mkdtemp__custom( C_TEMPLATE = "abcdef" path = os_ops.mkdtemp(C_TEMPLATE) logging.info("Path is [{0}].".format(path)) - assert os.path.exists(path) - assert C_TEMPLATE in os.path.basename(path) - os.rmdir(path) - assert not os.path.exists(path) + LocalCheck.check_path_exists(os_ops, path) + assert os_ops.path_exists(path) + assert os_ops.isdir(path) + assert C_TEMPLATE in os_ops.get_path_basename(path) + os_ops.rmdir(path) + LocalCheck.check_path_does_not_exists(os_ops, path) + assert not os_ops.path_exists(path) return def test_rmdirs( @@ -622,11 +626,44 @@ def test_rmdirs( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - path = os_ops.mkdtemp() - assert os.path.exists(path) + tmpdir = os_ops.get_tempdir() + + path = os_ops.build_path( + tmpdir, + "testgres-os_ops-test_rmdirs-" + uuid.uuid4().bytes.hex(), + ) + + local_detecter_is_created = False + if OsOpsHelpers.is_localhost(os_ops): + pass + elif sys.platform != os_ops.get_platform(): + pass + elif not os.path.exists(tmpdir): + pass + else: + # We will check a real work with another host + assert not os.path.exists(path) + os.mkdir(path) + assert os.path.exists(path) + local_detecter_is_created = True + logging.info("Local detecter is created [{}]".format(path)) + + cmd = ["mkdir", path] + os_ops.exec_command(cmd) + + LocalCheck.check_path_exists(os_ops, path) + assert os_ops.path_exists(path) + assert os_ops.isdir(path) assert os_ops.rmdirs(path, ignore_errors=False) is True - assert not os.path.exists(path) + LocalCheck.check_path_does_not_exists(os_ops, path) + assert not os_ops.path_exists(path) + + if local_detecter_is_created: + assert os.path.exists(path) + os.rmdir(path) + logging.info("Local detecter is deleted [{}]".format(path)) + return def test_rmdirs__01_with_subfolder( @@ -641,17 +678,23 @@ def test_rmdirs__01_with_subfolder( # folder with subfolder path = os_ops.mkdtemp() - assert os.path.exists(path) + LocalCheck.check_path_exists(os_ops, path) + assert os_ops.path_exists(path) - dir1 = os.path.join(path, "dir1") - assert not os.path.exists(dir1) + dir1 = os_ops.build_path(path, "dir1") + LocalCheck.check_path_does_not_exists(os_ops, dir1) + assert not os_ops.path_exists(dir1) - os_ops.makedirs(dir1) - assert os.path.exists(dir1) + os_ops.makedir(dir1) + LocalCheck.check_path_exists(os_ops, dir1) + assert os_ops.path_exists(dir1) + assert os_ops.isdir(dir1) assert os_ops.rmdirs(path, ignore_errors=False) is True - assert not os.path.exists(path) - assert not os.path.exists(dir1) + LocalCheck.check_path_does_not_exists(os_ops, path) + LocalCheck.check_path_does_not_exists(os_ops, dir1) + assert not os_ops.path_exists(path) + assert not os_ops.path_exists(dir1) return def test_rmdirs__02_with_file( @@ -666,17 +709,23 @@ def test_rmdirs__02_with_file( # folder with file path = os_ops.mkdtemp() - assert os.path.exists(path) + LocalCheck.check_path_exists(os_ops, path) + assert os_ops.path_exists(path) - file1 = os.path.join(path, "file1.txt") - assert not os.path.exists(file1) + file1 = os_ops.build_path(path, "file1.txt") + LocalCheck.check_path_does_not_exists(os_ops, file1) + assert not os_ops.path_exists(file1) os_ops.touch(file1) - assert os.path.exists(file1) + LocalCheck.check_path_exists(os_ops, file1) + assert os_ops.path_exists(file1) + assert os_ops.isfile(file1) assert os_ops.rmdirs(path, ignore_errors=False) is True - assert not os.path.exists(path) - assert not os.path.exists(file1) + LocalCheck.check_path_does_not_exists(os_ops, path) + LocalCheck.check_path_does_not_exists(os_ops, file1) + assert not os_ops.path_exists(path) + assert not os_ops.path_exists(file1) return def test_rmdirs__03_with_subfolder_and_file( @@ -691,24 +740,36 @@ def test_rmdirs__03_with_subfolder_and_file( # folder with subfolder and file path = os_ops.mkdtemp() - assert os.path.exists(path) + LocalCheck.check_path_exists(os_ops, path) + assert os_ops.path_exists(path) - dir1 = os.path.join(path, "dir1") - assert not os.path.exists(dir1) + dir1 = os_ops.build_path(path, "dir1") + LocalCheck.check_path_does_not_exists(os_ops, dir1) + assert not os_ops.path_exists(dir1) os_ops.makedirs(dir1) - assert os.path.exists(dir1) + LocalCheck.check_path_exists(os_ops, dir1) + assert os_ops.path_exists(dir1) + assert os_ops.isdir(dir1) + assert not os_ops.isfile(dir1) - file1 = os.path.join(dir1, "file1.txt") - assert not os.path.exists(file1) + file1 = os_ops.build_path(dir1, "file1.txt") + LocalCheck.check_path_does_not_exists(os_ops, file1) + assert not os_ops.path_exists(file1) os_ops.touch(file1) - assert os.path.exists(file1) + LocalCheck.check_path_exists(os_ops, file1) + assert os_ops.path_exists(file1) + assert os_ops.isfile(file1) + assert not os_ops.isdir(file1) assert os_ops.rmdirs(path, ignore_errors=False) is True - assert not os.path.exists(path) - assert not os.path.exists(dir1) - assert not os.path.exists(file1) + LocalCheck.check_path_does_not_exists(os_ops, path) + LocalCheck.check_path_does_not_exists(os_ops, dir1) + LocalCheck.check_path_does_not_exists(os_ops, file1) + assert not os_ops.path_exists(path) + assert not os_ops.path_exists(dir1) + assert not os_ops.path_exists(file1) return def test_write_text_file( @@ -821,13 +882,21 @@ def test_read__text( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - filename = __file__ # current file - - with open(filename, 'r') as file: # open in a text mode + with open(__file__, 'r', encoding="utf-8") as file: response0 = file.read() assert type(response0) is str + filename = os_ops.mkstemp( + "testgres-os_ops-test_read__text", + ) + + os_ops.write( + filename, + response0, + binary=False, + ) + response1 = os_ops.read(filename) assert type(response1) is str assert response1 == response0 @@ -843,6 +912,8 @@ def test_read__text( response4 = os_ops.read(filename, encoding="UTF-8") assert type(response4) is str assert response4 == response0 + + os_ops.remove_file(filename) return def test_read__binary( @@ -858,16 +929,26 @@ def test_read__binary( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - filename = __file__ # current file - - with open(filename, 'rb') as file: # open in a binary mode + with open(__file__, 'rb') as file: response0 = file.read() assert type(response0) is bytes + filename = os_ops.mkstemp( + "testgres-os_ops-test_read__binary", + ) + + os_ops.write( + filename, + response0, + binary=True, + ) + response1 = os_ops.read(filename, binary=True) assert type(response1) is bytes assert response1 == response0 + + os_ops.remove_file(filename) return def test_read__binary_and_encoding( @@ -904,13 +985,21 @@ def test_read_binary__spec( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - filename = __file__ # currnt file - - with open(filename, 'rb') as file: # open in a binary mode + with open(__file__, 'rb') as file: response0 = file.read() assert type(response0) is bytes + filename = os_ops.mkstemp( + "testgres-os_ops-test_read_binary__spec", + ) + + os_ops.write( + filename, + response0, + binary=True, + ) + response1 = os_ops.read_binary(filename, 0) assert type(response1) is bytes assert response1 == response0 @@ -933,6 +1022,8 @@ def test_read_binary__spec( response5 = os_ops.read_binary(filename, len(response1) + 1) assert type(response5) is bytes assert len(response5) == 0 + + os_ops.remove_file(filename) return def test_read_binary__spec__negative_offset( @@ -967,14 +1058,22 @@ def test_get_file_size( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - filename = __file__ # current file + filename = os_ops.mkstemp("testgres-os_ops-test_get_file_size-") + sz = os_ops.get_file_size(filename) + assert type(sz) is int + assert sz == 0 + + os_ops.write(filename, b"\x02\x01\x00", binary=True) + sz = os_ops.get_file_size(filename) + assert type(sz) is int + assert sz == 3 - sz0 = os.path.getsize(filename) - assert type(sz0) is int + os_ops.write(filename, b"\x04", binary=True, truncate=False) + sz = os_ops.get_file_size(filename) + assert type(sz) is int + assert sz == 4 - sz1 = os_ops.get_file_size(filename) - assert type(sz1) is int - assert sz1 == sz0 + os_ops.remove_file(filename) return def test_isfile_true( @@ -990,10 +1089,12 @@ def test_isfile_true( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - filename = __file__ + RunConditions.skip_if_windows() - response = os_ops.isfile(filename) + filename = "/bin/sh" + LocalCheck.check_isfile(os_ops, filename) + response = os_ops.isfile(filename) assert response is True return @@ -1010,11 +1111,41 @@ def test_isfile_false__not_exist( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - filename = os.path.join(os.path.dirname(__file__), "nonexistent_file.txt") + tmpdir = os_ops.get_tempdir() - response = os_ops.isfile(filename) + filename = os_ops.build_path( + tmpdir, + "nonexistent_file-{}.txt".format(uuid.uuid4().bytes.hex()), + ) + + LocalCheck.check_path_does_not_exists(os_ops, filename) + assert not os_ops.path_exists(filename) + + local_detecter_is_created = False + if OsOpsHelpers.is_localhost(os_ops): + pass + elif sys.platform != os_ops.get_platform(): + pass + elif not os.path.exists(tmpdir): + pass + else: + # We will check a real work with another host + assert not os.path.exists(filename) + with open(filename, "a"): + os.utime(filename, None) + assert os.path.exists(filename) + assert os.path.isfile(filename) + local_detecter_is_created = True + logging.info("Local detecter is created [{}]".format(filename)) + response = os_ops.isfile(filename) assert response is False + + if local_detecter_is_created: + assert os.path.exists(filename) + os.remove(filename) + assert not os.path.exists(filename) + logging.info("Local detecter is deleted [{}]".format(filename)) return def test_isfile_false__directory( @@ -1030,12 +1161,14 @@ def test_isfile_false__directory( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - name = os.path.dirname(__file__) - - assert os_ops.isdir(name) - - response = os_ops.isfile(name) + tmpdir = os_ops.get_tempdir() + LocalCheck.check_path_exists(os_ops, tmpdir) + LocalCheck.check_isdir(os_ops, tmpdir) + LocalCheck.check_not_isfile(os_ops, tmpdir) + assert os_ops.path_exists(tmpdir) + assert os_ops.isdir(tmpdir) + response = os_ops.isfile(tmpdir) assert response is False return @@ -1052,10 +1185,13 @@ def test_isdir_true( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - name = os.path.dirname(__file__) - - response = os_ops.isdir(name) + tmpdir = os_ops.get_tempdir() + LocalCheck.check_path_exists(os_ops, tmpdir) + LocalCheck.check_isdir(os_ops, tmpdir) + LocalCheck.check_not_isfile(os_ops, tmpdir) + assert os_ops.path_exists(tmpdir) + response = os_ops.isdir(tmpdir) assert response is True return @@ -1072,10 +1208,19 @@ def test_isdir_false__not_exist( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - name = os.path.join(os.path.dirname(__file__), "it_is_nonexistent_directory") + tmpdir = os_ops.get_tempdir() + LocalCheck.check_path_exists(os_ops, tmpdir) + LocalCheck.check_isdir(os_ops, tmpdir) + LocalCheck.check_not_isfile(os_ops, tmpdir) + assert os_ops.path_exists(tmpdir) + assert os_ops.isdir(tmpdir) is True - response = os_ops.isdir(name) + name = os_ops.build_path( + tmpdir, + "it_is_nonexistent_directory-{}".format(uuid.uuid4().bytes.hex()), + ) + response = os_ops.isdir(name) assert response is False return @@ -1092,13 +1237,19 @@ def test_isdir_false__file( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - name = __file__ - - assert os_ops.isfile(name) + name = os_ops.mkstemp() + LocalCheck.check_path_exists(os_ops, name) + LocalCheck.check_isfile(os_ops, name) + LocalCheck.check_not_isdir(os_ops, name) + assert os_ops.path_exists(name) is True + assert os_ops.isfile(name) is True response = os_ops.isdir(name) - assert response is False + + os_ops.remove_file(name) + LocalCheck.check_path_does_not_exists(os_ops, name) + assert os_ops.path_exists(name) is False return def test_cwd( @@ -1186,24 +1337,33 @@ def test_write( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - mode = "w+b" if write_data001.call_param__binary else "w+" - - with tempfile.NamedTemporaryFile(mode=mode, delete=True) as tmp_file: - tmp_file.write(write_data001.source) - tmp_file.flush() - - os_ops.write( - tmp_file.name, - write_data001.call_param__data, - read_and_write=write_data001.call_param__rw, - truncate=write_data001.call_param__truncate, - binary=write_data001.call_param__binary) + tmp_file = os_ops.mkstemp("testgres-os_ops-test_write") - tmp_file.seek(0) - - s = tmp_file.read() + os_ops.write( + tmp_file, + write_data001.source, + binary=write_data001.call_param__binary, + ) + s = os_ops.read( + tmp_file, + binary=write_data001.call_param__binary, + ) + assert s == write_data001.source + + os_ops.write( + tmp_file, + write_data001.call_param__data, + read_and_write=write_data001.call_param__rw, + truncate=write_data001.call_param__truncate, + binary=write_data001.call_param__binary, + ) + s = os_ops.read( + tmp_file, + binary=write_data001.call_param__binary, + ) + assert s == write_data001.result - assert s == write_data001.result + os_ops.remove_file(tmp_file) return def test_touch( @@ -1240,21 +1400,44 @@ def test_is_port_free__true( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - C_LIMIT = 128 + C_SAMPLES = 128 + C_LIMIT = 10 - ports = set(range(1024, 65535)) - assert type(ports) is set + ports = random.sample(range(1024, 65536), C_SAMPLES) + assert type(ports) is list ok_count = 0 no_count = 0 + py_test_code_templ = """ +import socket +with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + try: + s.bind(("", {0})) + except OSError: + exit(123) +print(str({0})) +exit(0) +""" for port in ports: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - try: - s.bind(("", port)) - except OSError: + logging.info("Try to bind port {}...".format(port)) + + py_test_code = py_test_code_templ.format(port) + + try: + r = os_ops.exec_command( + ["python3", "-c", py_test_code], + encoding="utf-8", + ) + except ExecUtilException as e: + if e.exit_code == 123: + logging.info("Fails") continue + raise + + assert r == str(port) + "\n" + logging.info("Try to check via is_port_free ...") r = os_ops.is_port_free(port) if r: @@ -1266,9 +1449,7 @@ def test_is_port_free__true( if ok_count == C_LIMIT: return - - if no_count == C_LIMIT: - raise RuntimeError("To many false positive test attempts.") + continue if ok_count == 0: raise RuntimeError("No one free port was found.") @@ -1284,60 +1465,70 @@ def test_is_port_free__false( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - C_LIMIT = 10 - - ports = set(range(1024, 65535)) - assert type(ports) is set + C_LIMIT = 5 + C_SAMPLES = C_LIMIT * 2 - def LOCAL_server(s: socket.socket): - assert s is not None - assert type(s) is socket.socket - - try: - while True: - r = s.accept() - - if r is None: - break - except Exception as e: - assert e is not None - pass + ports = random.sample(range(1024, 65536), C_SAMPLES) ok_count = 0 no_count = 0 + py_server_code_templ = """ +import socket, time +s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) +try: + s.bind(("", {0})) + s.listen(1) + print("READY", flush=True) + time.sleep(30) # Keep the port busy for 30 seconds +except OSError: + exit(123) +exit(0) +""" for port in ports: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - try: - s.bind(("", port)) - except OSError: - continue + logging.info("Try to occupy port {} on target machine...".format(port)) - th = threading.Thread(target=LOCAL_server, args=[s]) + py_server_code = py_server_code_templ.format(port) - s.listen(10) + # Start a background process on the target machine + p = os_ops.exec_command( + ["python3", "-u", "-c", py_server_code], + get_process=True, + encoding="utf-8", + ) + assert isinstance(p, subprocess.Popen) + assert p.stdout is not None - assert type(th) is threading.Thread - th.start() + try: + # Read the first line from the process's stdout. + # If it's READY, the socket was successfully bound. + # If the process crashed (code 123), readline() will return empty. + ready_line = p.stdout.readline() - try: - r = os_ops.is_port_free(port) - finally: - s.shutdown(2) - th.join() + if ready_line != "READY\n": + logging.info("Port {} is already busy or failed to bind, skipping...".format(port)) + continue # it jumps into finally + + logging.info("Port {} is occupied. Verifying via is_port_free...".format(port)) + + # MAIN CHECK: The port is currently busy, so is_port_free should return False! + r = os_ops.is_port_free(port) + assert type(r) is bool if not r: ok_count += 1 - logging.info("OK. Port {} is not free.".format(port)) + logging.info("OK. Port {} is correctly detected as NOT free.".format(port)) else: no_count += 1 - logging.warning("NO. Port {} does not accept connection.".format(port)) + logging.warning("NO. Port {} was detected as free, but it is busy!".format(port)) + finally: + # We guarantee that the process will be terminated and the port will be released on the target machine. + p.terminate() + p.wait() - if ok_count == C_LIMIT: - return - - if no_count == C_LIMIT: - raise RuntimeError("To many false positive test attempts.") + if ok_count == C_LIMIT: + return + continue if ok_count == 0: raise RuntimeError("No one free port was found.") @@ -1355,24 +1546,29 @@ def test_get_tempdir( dir = os_ops.get_tempdir() assert type(dir) is str - assert os_ops.path_exists(dir) - assert os.path.exists(dir) + LocalCheck.check_path_exists(os_ops, dir) + assert os_ops.path_exists(dir) is True + assert os_ops.isdir(dir) is True - file_path = os.path.join(dir, "testgres--" + uuid.uuid4().hex + ".tmp") + file_path = os_ops.build_path( + dir, + "testgres--" + uuid.uuid4().hex + ".tmp", + ) os_ops.write(file_path, "1234", binary=False) - assert os_ops.path_exists(file_path) - assert os.path.exists(file_path) + LocalCheck.check_path_exists(os_ops, file_path) + LocalCheck.check_isfile(os_ops, file_path) + assert os_ops.path_exists(file_path) is True + assert os_ops.isfile(file_path) is True + assert os_ops.get_file_size(file_path) == 4 d = os_ops.read(file_path, binary=False) - assert d == "1234" os_ops.remove_file(file_path) - - assert not os_ops.path_exists(file_path) - assert not os.path.exists(file_path) + LocalCheck.check_path_does_not_exists(os_ops, file_path) + assert os_ops.path_exists(file_path) is False return def test_get_tempdir__compare_with_py_info( @@ -1390,7 +1586,7 @@ def test_get_tempdir__compare_with_py_info( assert type(actual_dir) is str # -------- - cmd = [sys.executable, "-c", "import tempfile;print(tempfile.gettempdir());"] + cmd = ["python3", "-c", "import tempfile;print(tempfile.gettempdir());"] expected_dir_b = os_ops.exec_command(cmd) assert type(expected_dir_b) is bytes @@ -1440,12 +1636,14 @@ def test_mkdir__mt(self, data001: tagData_OS_OPS__NUMS): logging.info("A lock file [{}] is creating ...".format(lock_dir)) - assert os.path.exists(lock_dir) + LocalCheck.check_path_exists(os_ops, lock_dir) + assert os_ops.path_exists(lock_dir) is True - def MAKE_PATH(lock_dir: str, num: int) -> str: + def MAKE_PATH(os_ops: OsOperations, lock_dir: str, num: int) -> str: + assert isinstance(os_ops, OsOperations) assert type(lock_dir) is str assert type(num) is int - return os.path.join(lock_dir, str(num) + ".lock") + return os_ops.build_path(lock_dir, str(num) + ".lock") def LOCAL_WORKER(os_ops: OsOperations, workerID: int, @@ -1460,7 +1658,7 @@ def LOCAL_WORKER(os_ops: OsOperations, assert cNumbers > 0 assert len(reservedNumbers) == 0 - assert os.path.exists(lock_dir) + assert os_ops.path_exists(lock_dir) def LOG_INFO(template: str, *args) -> None: assert type(template) is str @@ -1477,7 +1675,7 @@ def LOG_INFO(template: str, *args) -> None: for num in range(cNumbers): assert num not in reservedNumbers - file_path = MAKE_PATH(lock_dir, num) + file_path = MAKE_PATH(os_ops, lock_dir, num) try: os_ops.makedir(file_path) @@ -1600,7 +1798,7 @@ class tadWorkerData: else: reservedNumbers[n] = i - file_path = MAKE_PATH(lock_dir, n) + file_path = MAKE_PATH(os_ops, lock_dir, n) if not os_ops.path_exists(file_path): nErrors += 1 logging.error("File {} is not found!".format(file_path)) @@ -1616,7 +1814,7 @@ class tadWorkerData: logging.error("Number {} is not reserved!".format(n)) continue - file_path = MAKE_PATH(lock_dir, n) + file_path = MAKE_PATH(os_ops, lock_dir, n) if not os_ops.path_exists(file_path): nErrors += 1 logging.error("File {} is not found!".format(file_path)) @@ -1633,7 +1831,7 @@ class tadWorkerData: )) for n in range(N_NUMBERS): - file_path = MAKE_PATH(lock_dir, n) + file_path = MAKE_PATH(os_ops, lock_dir, n) try: os_ops.rmdir(file_path) except Exception as e: @@ -1664,42 +1862,42 @@ class tadWorkerData: logging.info("Test is finished! Total error count is {}.".format(nErrors)) return - T_KILL_SIGNAL_DESCR = typing.Tuple[ - str, - typing.Union[int, os_signal.Signals], - str - ] + @dataclasses.dataclass + class T_KILL_SIGNAL_DESCR: + sign: str + signal: typing.Union[int, os_signal.Signals] + signal_num_s: str sm_kill_signal_ids: typing.List[T_KILL_SIGNAL_DESCR] = [ - ("SIGINT", os_signal.SIGINT, "2"), - # ("SIGQUIT", os_signal.SIGQUIT, "3"), # it creates coredump - ("SIGKILL", os_signal.SIGKILL, "9"), - ("SIGTERM", os_signal.SIGTERM, "15"), - ("2", 2, "2"), - # ("3", 3, "3"), # it creates coredump - ("9", 9, "9"), - ("15", 15, "15"), + T_KILL_SIGNAL_DESCR("SIGINT", os_signal.SIGINT, "2"), + # T_KILL_SIGNAL_DESCR("SIGQUIT", os_signal.SIGQUIT, "3"), # it creates coredump + T_KILL_SIGNAL_DESCR("SIGKILL", os_signal.SIGKILL, "9"), + T_KILL_SIGNAL_DESCR("SIGTERM", os_signal.SIGTERM, "15"), + T_KILL_SIGNAL_DESCR("2", 2, "2"), + # T_KILL_SIGNAL_DESCR("3", 3, "3"), # it creates coredump + T_KILL_SIGNAL_DESCR("9", 9, "9"), + T_KILL_SIGNAL_DESCR("15", 15, "15"), ] @pytest.fixture( params=sm_kill_signal_ids, - ids=["signal: {}".format(x[0]) for x in sm_kill_signal_ids], + ids=["signal: {}".format(x.sign) for x in sm_kill_signal_ids], ) def kill_signal_id( self, request: pytest.FixtureRequest, ) -> T_KILL_SIGNAL_DESCR: assert isinstance(request, pytest.FixtureRequest) - assert type(request.param) is tuple + assert type(request.param).__name__ == "T_KILL_SIGNAL_DESCR" return request.param def test_kill_signal( self, kill_signal_id: T_KILL_SIGNAL_DESCR, ): - assert type(kill_signal_id) is tuple - assert "{}".format(kill_signal_id[1]) == kill_signal_id[2] - assert "{}".format(int(kill_signal_id[1])) == kill_signal_id[2] + assert type(kill_signal_id) is __class__.T_KILL_SIGNAL_DESCR + assert "{}".format(kill_signal_id.signal) == kill_signal_id.signal_num_s + assert "{}".format(int(kill_signal_id.signal)) == kill_signal_id.signal_num_s return def test_kill( @@ -1712,36 +1910,43 @@ def test_kill( """ assert type(os_ops_descr) is OsOpsDescr assert isinstance(os_ops_descr.os_ops, OsOperations) - assert type(kill_signal_id) is tuple + assert type(kill_signal_id) is __class__.T_KILL_SIGNAL_DESCR os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) cmd = [ - sys.executable, + "python3", + "-u", "-c", - "import time; print('ENTER');time.sleep(300);print('EXIT')" + "import os, time; print(os.getpid());time.sleep(300);print('EXIT')" ] logging.info("Local test process is creating ...") - proc = subprocess.Popen( + proc = os_ops.exec_command( cmd, - text=True, + encoding="utf-8", + get_process=True, ) assert proc is not None assert type(proc) is subprocess.Popen - proc_pid = proc.pid + assert proc.stdout is not None + line = proc.stdout.readline() + assert line is not None + assert type(line) is str + logging.info("proc output: {!r}".format(line)) + line = line.rstrip() + assert line != "" + proc_pid = int(line) assert type(proc_pid) is int logging.info("Test process pid is {}".format(proc_pid)) - logging.info("Get this test process ...") - p1 = psutil.Process(proc_pid) - assert p1 is not None - del p1 + logging.info("Check this test process ...") + assert os_ops.get_process_children(proc_pid) == [] logging.info("Kill this test process ...") - os_ops.kill(proc_pid, kill_signal_id[1]) + os_ops.kill(proc_pid, kill_signal_id.signal) logging.info("Wait for finish ...") proc.wait() @@ -1760,19 +1965,34 @@ def test_kill( time.sleep(1) try: - psutil.Process(proc_pid) - except psutil.ZombieProcess as e: - logging.info("Exception {}: {}".format( - type(e).__name__, - str(e), - )) - except psutil.NoSuchProcess: - logging.info("OK. Process died.") - break + os_ops.get_process_children(proc_pid) + except Exception as e: + if type(os_ops).__name__ == "LocalOperations": + if isinstance(e, psutil.ZombieProcess): + logging.info("Exception {}: {}".format( + type(e).__name__, + str(e), + )) + break + if isinstance(e, psutil.NoSuchProcess): + logging.info("OK. Process died.") + break + raise - logging.info("Process is alive!") - continue + if type(os_ops).__name__ == "RemoteOperations": + if isinstance(e, ExecUtilException): + assert e.exit_code == 1 + logging.info("OK. Process died.") + break + raise + logging.error("Unknown os_ops object: {}".format( + type(os_ops).__name__, + )) + raise + else: + logging.info("Process is alive!") + continue return def test_kill__unk_pid( @@ -1785,38 +2005,59 @@ def test_kill__unk_pid( """ assert type(os_ops_descr) is OsOpsDescr assert isinstance(os_ops_descr.os_ops, OsOperations) - assert type(kill_signal_id) is tuple + assert type(kill_signal_id) is __class__.T_KILL_SIGNAL_DESCR os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) cmd = [ - sys.executable, + "python3", + "-u", "-c", - "import sys; print(\"a\", file=sys.stdout); print(\"b\", file=sys.stderr)" + """ +import os, sys +print('a:' + str(os.getpid()), file=sys.stdout) +print('a2', file=sys.stdout) +print('b', file=sys.stderr) +""" ] logging.info("Local test process is creating ...") - proc = subprocess.Popen( + proc = os_ops.exec_command( cmd, - text=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, + encoding="utf-8", + get_process=True, ) assert proc is not None assert type(proc) is subprocess.Popen + proc_pid = proc.pid assert type(proc_pid) is int + assert proc.stdout is not None + line = proc.stdout.readline() + assert line is not None + assert type(line) is str + logging.info("proc output: {!r}".format(line)) + line = line.rstrip() + assert line != "" + assert line.startswith("a:") + line = line[2:] + assert line != "" + proc_pid = int(line) + assert type(proc_pid) is int logging.info("Test process pid is {}".format(proc_pid)) logging.info("Wait for finish ...") - pout, perr = proc.communicate() + pout = proc.stdout.read() + assert proc.stderr is not None + perr = proc.stderr.read() + proc.wait() logging.info("STDOUT: {}".format(pout)) - logging.info("STDERR: {}".format(pout)) + logging.info("STDERR: {}".format(perr)) assert type(pout) is str assert type(perr) is str - assert pout == "a\n" + assert pout == "a2\n" assert perr == "b\n" assert type(proc.returncode) is int assert proc.returncode == 0 @@ -1835,22 +2076,38 @@ def test_kill__unk_pid( time.sleep(1) try: - psutil.Process(proc_pid) - except psutil.ZombieProcess as e: - logging.info("Exception {}: {}".format( - type(e).__name__, - str(e), - )) - except psutil.NoSuchProcess: - logging.info("OK. Process died.") - break + os_ops.get_process_children(proc_pid) + except Exception as e: + if type(os_ops).__name__ == "LocalOperations": + if isinstance(e, psutil.ZombieProcess): + logging.info("Exception {}: {}".format( + type(e).__name__, + str(e), + )) + break + if isinstance(e, psutil.NoSuchProcess): + logging.info("OK. Process died.") + break + raise - logging.info("Process is alive!") - continue + if type(os_ops).__name__ == "RemoteOperations": + if isinstance(e, ExecUtilException): + assert e.exit_code == 1 + logging.info("OK. Process died.") + break + raise + + logging.error("Unknown os_ops object: {}".format( + type(os_ops).__name__, + )) + raise + else: + logging.info("Process is alive!") + continue # -------------------- with pytest.raises(expected_exception=Exception) as x: - os_ops.kill(proc_pid, kill_signal_id[1]) + os_ops.kill(proc_pid, kill_signal_id.signal) assert x is not None assert isinstance(x.value, Exception) @@ -1859,14 +2116,14 @@ def test_kill__unk_pid( logging.info("Our error is [{}]".format(str(x.value))) logging.info("Our exception has type [{}]".format(type(x.value).__name__)) - if type(os_ops).__name__ == "LocalOsOperations": + if type(os_ops).__name__ == "LocalOperations": assert type(x.value) is ProcessLookupError assert "No such process" in str(x.value) - elif type(os_ops).__name__ == "RemoteOsOperations": + elif type(os_ops).__name__ == "RemoteOperations": assert type(x.value) is ExecUtilException assert "No such process" in str(x.value) else: - RuntimeError("Unknown os_ops type: {}".format(type(os_ops).__name__)) + __class__.helper__bug_check__unknown_os_ops_type(os_ops) return @@ -1880,19 +2137,15 @@ def test_get_dirname( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - p = __file__ + expected_dirname = "abc" + + p = os_ops.build_path(expected_dirname, "file1.txt") assert type(p) is str assert p != "" - assert os.path.exists(p) - - expected_dirname = os.path.dirname(p) - assert type(expected_dirname) is str - assert expected_dirname != "" actual_dirname = os_ops.get_dirname(p) assert type(actual_dirname) is str assert actual_dirname != "" - assert actual_dirname == expected_dirname return @@ -1906,14 +2159,17 @@ def test_is_abs_path__yes( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - p = __file__ + p = os_ops.get_tempdir() assert type(p) is str assert p != "" - assert os.path.isabs(p) + LocalCheck.check_path_exists(os_ops, p) + LocalCheck.check_isdir(os_ops, p) + LocalCheck.check_path_is_abs(os_ops, p) + assert os_ops.path_exists(p) is True + assert os_ops.isdir(p) is True actual_value = os_ops.is_abs_path(p) assert type(actual_value) is bool - assert actual_value is True return @@ -1926,10 +2182,10 @@ def test_is_abs_path__no(self, os_ops_descr: OsOpsDescr): p = "." assert not os.path.isabs(p) + LocalCheck.check_path_is_not_abs(os_ops, p) actual_value = os_ops.is_abs_path(p) assert type(actual_value) is bool - assert actual_value is False return diff --git a/tests/test_os_ops_remote.py b/tests/test_os_ops_remote.py index b2da160..4a79490 100755 --- a/tests/test_os_ops_remote.py +++ b/tests/test_os_ops_remote.py @@ -3,10 +3,10 @@ from tests.helpers.global_data import OsOpsDescr from tests.helpers.global_data import OsOpsDescrs from tests.helpers.global_data import OsOperations +from tests.helpers.local_check import LocalCheck from src.exceptions import ExecUtilException -import os import pytest @@ -43,12 +43,14 @@ def test_rmdirs__try_to_delete_file( path = os_ops.mkstemp() assert type(path) is str - assert os.path.exists(path) + LocalCheck.check_path_exists(os_ops, path) + assert os_ops.path_exists(path) with pytest.raises(ExecUtilException) as x: os_ops.rmdirs(path, ignore_errors=False) - assert os.path.exists(path) + LocalCheck.check_path_exists(os_ops, path) + assert os_ops.path_exists(path) assert type(x.value) is ExecUtilException assert type(x.value.description) is str assert x.value.description == "Utility exited with non-zero code (20). Error: `cannot remove '" + path + "': it is not a directory`"