From 75b080484fff382725866697785059eb0d618876 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 08:57:28 +0300 Subject: [PATCH 01/11] TestOsOpsCommon is added [generic os_ops tests] --- tests/test_local.py | 221 ---------------------------- tests/test_os_ops_common.py | 278 ++++++++++++++++++++++++++++++++++++ tests/test_remote.py | 231 ------------------------------ 3 files changed, 278 insertions(+), 452 deletions(-) create mode 100644 tests/test_os_ops_common.py diff --git a/tests/test_local.py b/tests/test_local.py index 3ae93f76..684f0f46 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -3,11 +3,9 @@ import pytest import re -import tempfile import logging from ..testgres import ExecUtilException -from ..testgres import InvalidOperationException from ..testgres import LocalOperations from .helpers.run_conditions import RunConditions @@ -98,59 +96,6 @@ def test_listdir(self): assert f is not None assert type(f) == str # noqa: E721 - def test_read__text(self): - """ - Test LocalOperations::read for text data. - """ - filename = __file__ # current file - - with open(filename, 'r') as file: # open in a text mode - response0 = file.read() - - assert type(response0) == str # noqa: E721 - - response1 = self.operations.read(filename) - assert type(response1) == str # noqa: E721 - assert response1 == response0 - - response2 = self.operations.read(filename, encoding=None, binary=False) - assert type(response2) == str # noqa: E721 - assert response2 == response0 - - response3 = self.operations.read(filename, encoding="") - assert type(response3) == str # noqa: E721 - assert response3 == response0 - - response4 = self.operations.read(filename, encoding="UTF-8") - assert type(response4) == str # noqa: E721 - assert response4 == response0 - - def test_read__binary(self): - """ - Test LocalOperations::read for binary data. - """ - filename = __file__ # current file - - with open(filename, 'rb') as file: # open in a binary mode - response0 = file.read() - - assert type(response0) == bytes # noqa: E721 - - response1 = self.operations.read(filename, binary=True) - assert type(response1) == bytes # noqa: E721 - assert response1 == response0 - - def test_read__binary_and_encoding(self): - """ - Test LocalOperations::read for binary data and encoding. - """ - filename = __file__ # current file - - with pytest.raises( - InvalidOperationException, - match=re.escape("Enconding is not allowed for read binary operation")): - self.operations.read(filename, encoding="", binary=True) - def test_read__unknown_file(self): """ Test LocalOperations::read with unknown file. @@ -159,40 +104,6 @@ def test_read__unknown_file(self): with pytest.raises(FileNotFoundError, match=re.escape("[Errno 2] No such file or directory: '/dummy'")): self.operations.read("/dummy") - def test_read_binary__spec(self): - """ - Test LocalOperations::read_binary. - """ - filename = __file__ # current file - - with open(filename, 'rb') as file: # open in a binary mode - response0 = file.read() - - assert type(response0) == bytes # noqa: E721 - - response1 = self.operations.read_binary(filename, 0) - assert type(response1) == bytes # noqa: E721 - assert response1 == response0 - - response2 = self.operations.read_binary(filename, 1) - assert type(response2) == bytes # noqa: E721 - assert len(response2) < len(response1) - assert len(response2) + 1 == len(response1) - assert response2 == response1[1:] - - response3 = self.operations.read_binary(filename, len(response1)) - assert type(response3) == bytes # noqa: E721 - assert len(response3) == 0 - - response4 = self.operations.read_binary(filename, len(response2)) - assert type(response4) == bytes # noqa: E721 - assert len(response4) == 1 - assert response4[0] == response1[len(response1) - 1] - - response5 = self.operations.read_binary(filename, len(response1) + 1) - assert type(response5) == bytes # noqa: E721 - assert len(response5) == 0 - def test_read_binary__spec__unk_file(self): """ Test LocalOperations::read_binary with unknown file. @@ -234,70 +145,6 @@ def test_get_file_size__unk_file(self): with pytest.raises(FileNotFoundError, match=re.escape("[Errno 2] No such file or directory: '/dummy'")): self.operations.get_file_size("/dummy") - def test_isfile_true(self): - """ - Test isfile for an existing file. - """ - filename = __file__ - - response = self.operations.isfile(filename) - - assert response is True - - def test_isfile_false__not_exist(self): - """ - Test isfile for a non-existing file. - """ - filename = os.path.join(os.path.dirname(__file__), "nonexistent_file.txt") - - response = self.operations.isfile(filename) - - assert response is False - - def test_isfile_false__directory(self): - """ - Test isfile for a firectory. - """ - name = os.path.dirname(__file__) - - assert self.operations.isdir(name) - - response = self.operations.isfile(name) - - assert response is False - - def test_isdir_true(self): - """ - Test isdir for an existing directory. - """ - name = os.path.dirname(__file__) - - response = self.operations.isdir(name) - - assert response is True - - def test_isdir_false__not_exist(self): - """ - Test isdir for a non-existing directory. - """ - name = os.path.join(os.path.dirname(__file__), "it_is_nonexistent_directory") - - response = self.operations.isdir(name) - - assert response is False - - def test_isdir_false__file(self): - """ - Test isdir for a file. - """ - name = __file__ - - assert self.operations.isfile(name) - - response = self.operations.isdir(name) - - assert response is False - def test_cwd(self): """ Test cwd. @@ -314,71 +161,3 @@ def test_cwd(self): # Comp result assert v == expectedValue - - class tagWriteData001: - def __init__(self, sign, source, cp_rw, cp_truncate, cp_binary, cp_data, result): - self.sign = sign - self.source = source - self.call_param__rw = cp_rw - self.call_param__truncate = cp_truncate - self.call_param__binary = cp_binary - self.call_param__data = cp_data - self.result = result - - sm_write_data001 = [ - tagWriteData001("A001", "1234567890", False, False, False, "ABC", "1234567890ABC"), - tagWriteData001("A002", b"1234567890", False, False, True, b"ABC", b"1234567890ABC"), - - tagWriteData001("B001", "1234567890", False, True, False, "ABC", "ABC"), - tagWriteData001("B002", "1234567890", False, True, False, "ABC1234567890", "ABC1234567890"), - tagWriteData001("B003", b"1234567890", False, True, True, b"ABC", b"ABC"), - tagWriteData001("B004", b"1234567890", False, True, True, b"ABC1234567890", b"ABC1234567890"), - - tagWriteData001("C001", "1234567890", True, False, False, "ABC", "1234567890ABC"), - tagWriteData001("C002", b"1234567890", True, False, True, b"ABC", b"1234567890ABC"), - - tagWriteData001("D001", "1234567890", True, True, False, "ABC", "ABC"), - tagWriteData001("D002", "1234567890", True, True, False, "ABC1234567890", "ABC1234567890"), - tagWriteData001("D003", b"1234567890", True, True, True, b"ABC", b"ABC"), - tagWriteData001("D004", b"1234567890", True, True, True, b"ABC1234567890", b"ABC1234567890"), - - tagWriteData001("E001", "\0001234567890\000", False, False, False, "\000ABC\000", "\0001234567890\000\000ABC\000"), - tagWriteData001("E002", b"\0001234567890\000", False, False, True, b"\000ABC\000", b"\0001234567890\000\000ABC\000"), - - tagWriteData001("F001", "a\nb\n", False, False, False, ["c", "d"], "a\nb\nc\nd\n"), - tagWriteData001("F002", b"a\nb\n", False, False, True, [b"c", b"d"], b"a\nb\nc\nd\n"), - - tagWriteData001("G001", "a\nb\n", False, False, False, ["c\n\n", "d\n"], "a\nb\nc\nd\n"), - tagWriteData001("G002", b"a\nb\n", False, False, True, [b"c\n\n", b"d\n"], b"a\nb\nc\nd\n"), - ] - - @pytest.fixture( - params=sm_write_data001, - ids=[x.sign for x in sm_write_data001], - ) - def write_data001(self, request): - assert isinstance(request, pytest.FixtureRequest) - assert type(request.param) == __class__.tagWriteData001 # noqa: E721 - return request.param - - def test_write(self, write_data001): - assert type(write_data001) == __class__.tagWriteData001 # noqa: E721 - - 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() - - self.operations.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.seek(0) - - s = tmp_file.read() - - assert s == write_data001.result diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py new file mode 100644 index 00000000..12052112 --- /dev/null +++ b/tests/test_os_ops_common.py @@ -0,0 +1,278 @@ +# coding: utf-8 +from .helpers.os_ops_descrs import OsOpsDescr +from .helpers.os_ops_descrs import OsOpsDescrs +from .helpers.os_ops_descrs import OsOperations + +import os + +import pytest +import re +import tempfile + +from ..testgres import InvalidOperationException + + +class TestOsOpsCommon: + sm_os_ops_descrs: list[OsOpsDescr] = [ + OsOpsDescrs.sm_local_os_ops_descr, + OsOpsDescrs.sm_remote_os_ops_descr + ] + + @pytest.fixture( + params=[descr.os_ops for descr in sm_os_ops_descrs], + ids=[descr.sign for descr in sm_os_ops_descrs] + ) + def os_ops(self, request: pytest.FixtureRequest) -> OsOperations: + assert isinstance(request, pytest.FixtureRequest) + assert isinstance(request.param, OsOperations) + return request.param + + def test_read__text(self, os_ops: OsOperations): + """ + Test OsOperations::read for text data. + """ + assert isinstance(os_ops, OsOperations) + + filename = __file__ # current file + + with open(filename, 'r') as file: # open in a text mode + response0 = file.read() + + assert type(response0) == str # noqa: E721 + + response1 = os_ops.read(filename) + assert type(response1) == str # noqa: E721 + assert response1 == response0 + + response2 = os_ops.read(filename, encoding=None, binary=False) + assert type(response2) == str # noqa: E721 + assert response2 == response0 + + response3 = os_ops.read(filename, encoding="") + assert type(response3) == str # noqa: E721 + assert response3 == response0 + + response4 = os_ops.read(filename, encoding="UTF-8") + assert type(response4) == str # noqa: E721 + assert response4 == response0 + + def test_read__binary(self, os_ops: OsOperations): + """ + Test OsOperations::read for binary data. + """ + filename = __file__ # current file + + with open(filename, 'rb') as file: # open in a binary mode + response0 = file.read() + + assert type(response0) == bytes # noqa: E721 + + response1 = os_ops.read(filename, binary=True) + assert type(response1) == bytes # noqa: E721 + assert response1 == response0 + + def test_read__binary_and_encoding(self, os_ops: OsOperations): + """ + Test RemoteOperations::read for binary data and encoding. + """ + assert isinstance(os_ops, OsOperations) + + filename = __file__ # current file + + with pytest.raises( + InvalidOperationException, + match=re.escape("Enconding is not allowed for read binary operation")): + os_ops.read(filename, encoding="", binary=True) + + def test_read_binary__spec(self, os_ops: OsOperations): + """ + Test RemoteOperations::read_binary. + """ + assert isinstance(os_ops, OsOperations) + + filename = __file__ # currnt file + + with open(filename, 'rb') as file: # open in a binary mode + response0 = file.read() + + assert type(response0) == bytes # noqa: E721 + + response1 = os_ops.read_binary(filename, 0) + assert type(response1) == bytes # noqa: E721 + assert response1 == response0 + + response2 = os_ops.read_binary(filename, 1) + assert type(response2) == bytes # noqa: E721 + assert len(response2) < len(response1) + assert len(response2) + 1 == len(response1) + assert response2 == response1[1:] + + response3 = os_ops.read_binary(filename, len(response1)) + assert type(response3) == bytes # noqa: E721 + assert len(response3) == 0 + + response4 = os_ops.read_binary(filename, len(response2)) + assert type(response4) == bytes # noqa: E721 + assert len(response4) == 1 + assert response4[0] == response1[len(response1) - 1] + + response5 = os_ops.read_binary(filename, len(response1) + 1) + assert type(response5) == bytes # noqa: E721 + assert len(response5) == 0 + + def test_isfile_true(self, os_ops: OsOperations): + """ + Test isfile for an existing file. + """ + assert isinstance(os_ops, OsOperations) + + filename = __file__ + + response = os_ops.isfile(filename) + + assert response is True + + def test_isfile_false__not_exist(self, os_ops: OsOperations): + """ + Test isfile for a non-existing file. + """ + assert isinstance(os_ops, OsOperations) + + filename = os.path.join(os.path.dirname(__file__), "nonexistent_file.txt") + + response = os_ops.isfile(filename) + + assert response is False + + def test_isfile_false__directory(self, os_ops: OsOperations): + """ + Test isfile for a firectory. + """ + assert isinstance(os_ops, OsOperations) + + name = os.path.dirname(__file__) + + assert os_ops.isdir(name) + + response = os_ops.isfile(name) + + assert response is False + + def test_isdir_true(self, os_ops: OsOperations): + """ + Test isdir for an existing directory. + """ + assert isinstance(os_ops, OsOperations) + + name = os.path.dirname(__file__) + + response = os_ops.isdir(name) + + assert response is True + + def test_isdir_false__not_exist(self, os_ops: OsOperations): + """ + Test isdir for a non-existing directory. + """ + assert isinstance(os_ops, OsOperations) + + name = os.path.join(os.path.dirname(__file__), "it_is_nonexistent_directory") + + response = os_ops.isdir(name) + + assert response is False + + def test_isdir_false__file(self, os_ops: OsOperations): + """ + Test isdir for a file. + """ + assert isinstance(os_ops, OsOperations) + + name = __file__ + + assert os_ops.isfile(name) + + response = os_ops.isdir(name) + + assert response is False + + def test_cwd(self, os_ops: OsOperations): + """ + Test cwd. + """ + assert isinstance(os_ops, OsOperations) + + v = os_ops.cwd() + + assert v is not None + assert type(v) == str # noqa: E721 + assert v != "" + + class tagWriteData001: + def __init__(self, sign, source, cp_rw, cp_truncate, cp_binary, cp_data, result): + self.sign = sign + self.source = source + self.call_param__rw = cp_rw + self.call_param__truncate = cp_truncate + self.call_param__binary = cp_binary + self.call_param__data = cp_data + self.result = result + + sm_write_data001 = [ + tagWriteData001("A001", "1234567890", False, False, False, "ABC", "1234567890ABC"), + tagWriteData001("A002", b"1234567890", False, False, True, b"ABC", b"1234567890ABC"), + + tagWriteData001("B001", "1234567890", False, True, False, "ABC", "ABC"), + tagWriteData001("B002", "1234567890", False, True, False, "ABC1234567890", "ABC1234567890"), + tagWriteData001("B003", b"1234567890", False, True, True, b"ABC", b"ABC"), + tagWriteData001("B004", b"1234567890", False, True, True, b"ABC1234567890", b"ABC1234567890"), + + tagWriteData001("C001", "1234567890", True, False, False, "ABC", "1234567890ABC"), + tagWriteData001("C002", b"1234567890", True, False, True, b"ABC", b"1234567890ABC"), + + tagWriteData001("D001", "1234567890", True, True, False, "ABC", "ABC"), + tagWriteData001("D002", "1234567890", True, True, False, "ABC1234567890", "ABC1234567890"), + tagWriteData001("D003", b"1234567890", True, True, True, b"ABC", b"ABC"), + tagWriteData001("D004", b"1234567890", True, True, True, b"ABC1234567890", b"ABC1234567890"), + + tagWriteData001("E001", "\0001234567890\000", False, False, False, "\000ABC\000", "\0001234567890\000\000ABC\000"), + tagWriteData001("E002", b"\0001234567890\000", False, False, True, b"\000ABC\000", b"\0001234567890\000\000ABC\000"), + + tagWriteData001("F001", "a\nb\n", False, False, False, ["c", "d"], "a\nb\nc\nd\n"), + tagWriteData001("F002", b"a\nb\n", False, False, True, [b"c", b"d"], b"a\nb\nc\nd\n"), + + tagWriteData001("G001", "a\nb\n", False, False, False, ["c\n\n", "d\n"], "a\nb\nc\nd\n"), + tagWriteData001("G002", b"a\nb\n", False, False, True, [b"c\n\n", b"d\n"], b"a\nb\nc\nd\n"), + ] + + @pytest.fixture( + params=sm_write_data001, + ids=[x.sign for x in sm_write_data001], + ) + def write_data001(self, request): + assert isinstance(request, pytest.FixtureRequest) + assert type(request.param) == __class__.tagWriteData001 # noqa: E721 + return request.param + + def test_write(self, write_data001: tagWriteData001, os_ops: OsOperations): + assert type(write_data001) == __class__.tagWriteData001 # noqa: E721 + 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.seek(0) + + s = tmp_file.read() + + assert s == write_data001.result diff --git a/tests/test_remote.py b/tests/test_remote.py index 2c37e2c1..1591196f 100755 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -3,11 +3,9 @@ import pytest import re -import tempfile import logging from ..testgres import ExecUtilException -from ..testgres import InvalidOperationException from ..testgres import RemoteOperations from ..testgres import LocalOperations from ..testgres import ConnectionParams @@ -298,59 +296,6 @@ def test_read_binary_file(self): assert isinstance(response, bytes) - def test_read__text(self): - """ - Test RemoteOperations::read for text data. - """ - filename = __file__ # current file - - with open(filename, 'r') as file: # open in a text mode - response0 = file.read() - - assert type(response0) == str # noqa: E721 - - response1 = self.operations.read(filename) - assert type(response1) == str # noqa: E721 - assert response1 == response0 - - response2 = self.operations.read(filename, encoding=None, binary=False) - assert type(response2) == str # noqa: E721 - assert response2 == response0 - - response3 = self.operations.read(filename, encoding="") - assert type(response3) == str # noqa: E721 - assert response3 == response0 - - response4 = self.operations.read(filename, encoding="UTF-8") - assert type(response4) == str # noqa: E721 - assert response4 == response0 - - def test_read__binary(self): - """ - Test RemoteOperations::read for binary data. - """ - filename = __file__ # current file - - with open(filename, 'rb') as file: # open in a binary mode - response0 = file.read() - - assert type(response0) == bytes # noqa: E721 - - response1 = self.operations.read(filename, binary=True) - assert type(response1) == bytes # noqa: E721 - assert response1 == response0 - - def test_read__binary_and_encoding(self): - """ - Test RemoteOperations::read for binary data and encoding. - """ - filename = __file__ # current file - - with pytest.raises( - InvalidOperationException, - match=re.escape("Enconding is not allowed for read binary operation")): - self.operations.read(filename, encoding="", binary=True) - def test_read__unknown_file(self): """ Test RemoteOperations::read with unknown file. @@ -363,40 +308,6 @@ def test_read__unknown_file(self): assert "No such file or directory" in str(x.value) assert "/dummy" in str(x.value) - def test_read_binary__spec(self): - """ - Test RemoteOperations::read_binary. - """ - filename = __file__ # currnt file - - with open(filename, 'rb') as file: # open in a binary mode - response0 = file.read() - - assert type(response0) == bytes # noqa: E721 - - response1 = self.operations.read_binary(filename, 0) - assert type(response1) == bytes # noqa: E721 - assert response1 == response0 - - response2 = self.operations.read_binary(filename, 1) - assert type(response2) == bytes # noqa: E721 - assert len(response2) < len(response1) - assert len(response2) + 1 == len(response1) - assert response2 == response1[1:] - - response3 = self.operations.read_binary(filename, len(response1)) - assert type(response3) == bytes # noqa: E721 - assert len(response3) == 0 - - response4 = self.operations.read_binary(filename, len(response2)) - assert type(response4) == bytes # noqa: E721 - assert len(response4) == 1 - assert response4[0] == response1[len(response1) - 1] - - response5 = self.operations.read_binary(filename, len(response1) + 1) - assert type(response5) == bytes # noqa: E721 - assert len(response5) == 0 - def test_read_binary__spec__unk_file(self): """ Test RemoteOperations::read_binary with unknown file. @@ -453,145 +364,3 @@ def test_touch(self): self.operations.touch(filename) assert self.operations.isfile(filename) - - def test_isfile_true(self): - """ - Test isfile for an existing file. - """ - filename = __file__ - - response = self.operations.isfile(filename) - - assert response is True - - def test_isfile_false__not_exist(self): - """ - Test isfile for a non-existing file. - """ - filename = os.path.join(os.path.dirname(__file__), "nonexistent_file.txt") - - response = self.operations.isfile(filename) - - assert response is False - - def test_isfile_false__directory(self): - """ - Test isfile for a firectory. - """ - name = os.path.dirname(__file__) - - assert self.operations.isdir(name) - - response = self.operations.isfile(name) - - assert response is False - - def test_isdir_true(self): - """ - Test isdir for an existing directory. - """ - name = os.path.dirname(__file__) - - response = self.operations.isdir(name) - - assert response is True - - def test_isdir_false__not_exist(self): - """ - Test isdir for a non-existing directory. - """ - name = os.path.join(os.path.dirname(__file__), "it_is_nonexistent_directory") - - response = self.operations.isdir(name) - - assert response is False - - def test_isdir_false__file(self): - """ - Test isdir for a file. - """ - name = __file__ - - assert self.operations.isfile(name) - - response = self.operations.isdir(name) - - assert response is False - - def test_cwd(self): - """ - Test cwd. - """ - v = self.operations.cwd() - - assert v is not None - assert type(v) == str # noqa: E721 - assert v != "" - - class tagWriteData001: - def __init__(self, sign, source, cp_rw, cp_truncate, cp_binary, cp_data, result): - self.sign = sign - self.source = source - self.call_param__rw = cp_rw - self.call_param__truncate = cp_truncate - self.call_param__binary = cp_binary - self.call_param__data = cp_data - self.result = result - - sm_write_data001 = [ - tagWriteData001("A001", "1234567890", False, False, False, "ABC", "1234567890ABC"), - tagWriteData001("A002", b"1234567890", False, False, True, b"ABC", b"1234567890ABC"), - - tagWriteData001("B001", "1234567890", False, True, False, "ABC", "ABC"), - tagWriteData001("B002", "1234567890", False, True, False, "ABC1234567890", "ABC1234567890"), - tagWriteData001("B003", b"1234567890", False, True, True, b"ABC", b"ABC"), - tagWriteData001("B004", b"1234567890", False, True, True, b"ABC1234567890", b"ABC1234567890"), - - tagWriteData001("C001", "1234567890", True, False, False, "ABC", "1234567890ABC"), - tagWriteData001("C002", b"1234567890", True, False, True, b"ABC", b"1234567890ABC"), - - tagWriteData001("D001", "1234567890", True, True, False, "ABC", "ABC"), - tagWriteData001("D002", "1234567890", True, True, False, "ABC1234567890", "ABC1234567890"), - tagWriteData001("D003", b"1234567890", True, True, True, b"ABC", b"ABC"), - tagWriteData001("D004", b"1234567890", True, True, True, b"ABC1234567890", b"ABC1234567890"), - - tagWriteData001("E001", "\0001234567890\000", False, False, False, "\000ABC\000", "\0001234567890\000\000ABC\000"), - tagWriteData001("E002", b"\0001234567890\000", False, False, True, b"\000ABC\000", b"\0001234567890\000\000ABC\000"), - - tagWriteData001("F001", "a\nb\n", False, False, False, ["c", "d"], "a\nb\nc\nd\n"), - tagWriteData001("F002", b"a\nb\n", False, False, True, [b"c", b"d"], b"a\nb\nc\nd\n"), - - tagWriteData001("G001", "a\nb\n", False, False, False, ["c\n\n", "d\n"], "a\nb\nc\nd\n"), - tagWriteData001("G002", b"a\nb\n", False, False, True, [b"c\n\n", b"d\n"], b"a\nb\nc\nd\n"), - ] - - @pytest.fixture( - params=sm_write_data001, - ids=[x.sign for x in sm_write_data001], - ) - def write_data001(self, request): - assert isinstance(request, pytest.FixtureRequest) - assert type(request.param) == __class__.tagWriteData001 # noqa: E721 - return request.param - - def test_write(self, write_data001): - assert type(write_data001) == __class__.tagWriteData001 # noqa: E721 - - 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() - - self.operations.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.seek(0) - - s = tmp_file.read() - - assert s == write_data001.result From c2c2d4cb6c139724936d3b38809681ae71a52eb8 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 10:22:56 +0300 Subject: [PATCH 02/11] TestOsOpsCommon is updated New tests: - test_read_binary__spec__negative_offset - test_get_file_size --- tests/test_local.py | 23 ----------------------- tests/test_os_ops_common.py | 26 ++++++++++++++++++++++++++ tests/test_remote.py | 23 ----------------------- 3 files changed, 26 insertions(+), 46 deletions(-) diff --git a/tests/test_local.py b/tests/test_local.py index 684f0f46..82d0c98e 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -114,29 +114,6 @@ def test_read_binary__spec__unk_file(self): match=re.escape("[Errno 2] No such file or directory: '/dummy'")): self.operations.read_binary("/dummy", 0) - def test_read_binary__spec__negative_offset(self): - """ - Test LocalOperations::read_binary with negative offset. - """ - - with pytest.raises( - ValueError, - match=re.escape("Negative 'offset' is not supported.")): - self.operations.read_binary(__file__, -1) - - def test_get_file_size(self): - """ - Test LocalOperations::get_file_size. - """ - filename = __file__ # current file - - sz0 = os.path.getsize(filename) - assert type(sz0) == int # noqa: E721 - - sz1 = self.operations.get_file_size(filename) - assert type(sz1) == int # noqa: E721 - assert sz1 == sz0 - def test_get_file_size__unk_file(self): """ Test LocalOperations::get_file_size. diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 12052112..0463067d 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -120,6 +120,32 @@ def test_read_binary__spec(self, os_ops: OsOperations): assert type(response5) == bytes # noqa: E721 assert len(response5) == 0 + def test_read_binary__spec__negative_offset(self, os_ops: OsOperations): + """ + Test RemoteOperations::read_binary with negative offset. + """ + assert isinstance(os_ops, OsOperations) + + with pytest.raises( + ValueError, + match=re.escape("Negative 'offset' is not supported.")): + os_ops.read_binary(__file__, -1) + + def test_get_file_size(self, os_ops: OsOperations): + """ + Test RemoteOperations::get_file_size. + """ + assert isinstance(os_ops, OsOperations) + + filename = __file__ # current file + + sz0 = os.path.getsize(filename) + assert type(sz0) == int # noqa: E721 + + sz1 = os_ops.get_file_size(filename) + assert type(sz1) == int # noqa: E721 + assert sz1 == sz0 + def test_isfile_true(self, os_ops: OsOperations): """ Test isfile for an existing file. diff --git a/tests/test_remote.py b/tests/test_remote.py index 1591196f..cd2b592d 100755 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -320,29 +320,6 @@ def test_read_binary__spec__unk_file(self): assert "No such file or directory" in str(x.value) assert "/dummy" in str(x.value) - def test_read_binary__spec__negative_offset(self): - """ - Test RemoteOperations::read_binary with negative offset. - """ - - with pytest.raises( - ValueError, - match=re.escape("Negative 'offset' is not supported.")): - self.operations.read_binary(__file__, -1) - - def test_get_file_size(self): - """ - Test RemoteOperations::get_file_size. - """ - filename = __file__ # current file - - sz0 = os.path.getsize(filename) - assert type(sz0) == int # noqa: E721 - - sz1 = self.operations.get_file_size(filename) - assert type(sz1) == int # noqa: E721 - assert sz1 == sz0 - def test_get_file_size__unk_file(self): """ Test RemoteOperations::get_file_size. From 6e45c150ad5b5e3fe832d274faef5029e97d84d2 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 10:41:04 +0300 Subject: [PATCH 03/11] TestOsOpsCommon is updated New tests: - test_listdir - test_path_exists_true__directory - test_path_exists_true__file - test_path_exists_false__directory - test_path_exists_false__file - test_write_text_file - test_write_binary_file - test_read_text_file - test_read_binary_file --- tests/test_local.py | 11 ---- tests/test_os_ops_common.py | 119 ++++++++++++++++++++++++++++++++++++ tests/test_remote.py | 83 ------------------------- 3 files changed, 119 insertions(+), 94 deletions(-) diff --git a/tests/test_local.py b/tests/test_local.py index 82d0c98e..1e7bfa11 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -85,17 +85,6 @@ def test_exec_command_failure__expect_error(self): assert b"nonexistent_command" in error assert b"not found" in error - def test_listdir(self): - """ - Test listdir for listing directory contents. - """ - path = "/etc" - files = self.operations.listdir(path) - assert isinstance(files, list) - for f in files: - assert f is not None - assert type(f) == str # noqa: E721 - def test_read__unknown_file(self): """ Test LocalOperations::read with unknown file. diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 0463067d..a1c6cf44 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -2,6 +2,7 @@ from .helpers.os_ops_descrs import OsOpsDescr from .helpers.os_ops_descrs import OsOpsDescrs from .helpers.os_ops_descrs import OsOperations +from .helpers.run_conditions import RunConditions import os @@ -27,6 +28,124 @@ def os_ops(self, request: pytest.FixtureRequest) -> OsOperations: assert isinstance(request.param, OsOperations) return request.param + def test_listdir(self, os_ops: OsOperations): + """ + Test listdir for listing directory contents. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + path = "/etc" + files = os_ops.listdir(path) + assert isinstance(files, list) + for f in files: + assert f is not None + assert type(f) == str # noqa: E721 + + def test_path_exists_true__directory(self, os_ops: OsOperations): + """ + Test path_exists for an existing directory. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + assert os_ops.path_exists("/etc") is True + + def test_path_exists_true__file(self, os_ops: OsOperations): + """ + Test path_exists for an existing file. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + assert os_ops.path_exists(__file__) is True + + def test_path_exists_false__directory(self, os_ops: OsOperations): + """ + Test path_exists for a non-existing directory. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + assert os_ops.path_exists("/nonexistent_path") is False + + def test_path_exists_false__file(self, os_ops: OsOperations): + """ + Test path_exists for a non-existing file. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + assert os_ops.path_exists("/etc/nonexistent_path.txt") is False + + def test_write_text_file(self, os_ops: OsOperations): + """ + Test write for writing data to a text file. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + filename = "/tmp/test_file.txt" + data = "Hello, world!" + + os_ops.write(filename, data, truncate=True) + os_ops.write(filename, data) + + response = os_ops.read(filename) + + assert response == data + data + + def test_write_binary_file(self, os_ops: OsOperations): + """ + Test write for writing data to a binary file. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + filename = "/tmp/test_file.bin" + data = b"\x00\x01\x02\x03" + + os_ops.write(filename, data, binary=True, truncate=True) + + response = os_ops.read(filename, binary=True) + + assert response == data + + def test_read_text_file(self, os_ops: OsOperations): + """ + Test read for reading data from a text file. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + filename = "/etc/hosts" + + response = os_ops.read(filename) + + assert isinstance(response, str) + + def test_read_binary_file(self, os_ops: OsOperations): + """ + Test read for reading data from a binary file. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + filename = "/usr/bin/python3" + + response = os_ops.read(filename, binary=True) + + assert isinstance(response, bytes) + def test_read__text(self, os_ops: OsOperations): """ Test OsOperations::read for text data. diff --git a/tests/test_remote.py b/tests/test_remote.py index cd2b592d..88c2ede3 100755 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -2,7 +2,6 @@ import os import pytest -import re import logging from ..testgres import ExecUtilException @@ -214,88 +213,6 @@ def test_rmdirs__try_to_delete_file(self): assert type(x.value.exit_code) == int # noqa: E721 assert x.value.exit_code == 20 - def test_listdir(self): - """ - Test listdir for listing directory contents. - """ - path = "/etc" - files = self.operations.listdir(path) - assert isinstance(files, list) - for f in files: - assert f is not None - assert type(f) == str # noqa: E721 - - def test_path_exists_true__directory(self): - """ - Test path_exists for an existing directory. - """ - assert self.operations.path_exists("/etc") is True - - def test_path_exists_true__file(self): - """ - Test path_exists for an existing file. - """ - assert self.operations.path_exists(__file__) is True - - def test_path_exists_false__directory(self): - """ - Test path_exists for a non-existing directory. - """ - assert self.operations.path_exists("/nonexistent_path") is False - - def test_path_exists_false__file(self): - """ - Test path_exists for a non-existing file. - """ - assert self.operations.path_exists("/etc/nonexistent_path.txt") is False - - def test_write_text_file(self): - """ - Test write for writing data to a text file. - """ - filename = "/tmp/test_file.txt" - data = "Hello, world!" - - self.operations.write(filename, data, truncate=True) - self.operations.write(filename, data) - - response = self.operations.read(filename) - - assert response == data + data - - def test_write_binary_file(self): - """ - Test write for writing data to a binary file. - """ - filename = "/tmp/test_file.bin" - data = b"\x00\x01\x02\x03" - - self.operations.write(filename, data, binary=True, truncate=True) - - response = self.operations.read(filename, binary=True) - - assert response == data - - def test_read_text_file(self): - """ - Test read for reading data from a text file. - """ - filename = "/etc/hosts" - - response = self.operations.read(filename) - - assert isinstance(response, str) - - def test_read_binary_file(self): - """ - Test read for reading data from a binary file. - """ - filename = "/usr/bin/python3" - - response = self.operations.read(filename, binary=True) - - assert isinstance(response, bytes) - def test_read__unknown_file(self): """ Test RemoteOperations::read with unknown file. From e43a38484ee4a82c0dff69f21fab7d8d6839d6da Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 11:06:20 +0300 Subject: [PATCH 04/11] TestOsOpsCommon is updated New tests: - test_exec_command_success - test_exec_command_failure - test_exec_command_failure__expect_error --- tests/test_local.py | 55 --------------------------------- tests/test_os_ops_common.py | 61 +++++++++++++++++++++++++++++++++++++ tests/test_remote.py | 46 ---------------------------- 3 files changed, 61 insertions(+), 101 deletions(-) diff --git a/tests/test_local.py b/tests/test_local.py index 1e7bfa11..9514f5f9 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -5,11 +5,8 @@ import re import logging -from ..testgres import ExecUtilException from ..testgres import LocalOperations -from .helpers.run_conditions import RunConditions - class TestLocalOperations: @@ -33,58 +30,6 @@ def test_mkdtemp__custom(self): os.rmdir(path) assert not os.path.exists(path) - def test_exec_command_success(self): - """ - Test exec_command for successful command execution. - """ - RunConditions.skip_if_windows() - - cmd = "python3 --version" - response = self.operations.exec_command(cmd, wait_exit=True, shell=True) - - assert b'Python 3.' in response - - def test_exec_command_failure(self): - """ - Test exec_command for command execution failure. - """ - RunConditions.skip_if_windows() - - cmd = "nonexistent_command" - while True: - try: - self.operations.exec_command(cmd, wait_exit=True, shell=True) - except ExecUtilException as e: - assert type(e.exit_code) == int # noqa: E721 - assert e.exit_code == 127 - - assert type(e.message) == str # noqa: E721 - assert type(e.error) == bytes # noqa: E721 - - assert e.message.startswith("Utility exited with non-zero code (127). Error:") - assert "nonexistent_command" in e.message - assert "not found" in e.message - assert b"nonexistent_command" in e.error - assert b"not found" in e.error - break - raise Exception("We wait an exception!") - - def test_exec_command_failure__expect_error(self): - """ - Test exec_command for command execution failure. - """ - RunConditions.skip_if_windows() - - cmd = "nonexistent_command" - - exit_status, result, error = self.operations.exec_command(cmd, verbose=True, wait_exit=True, shell=True, expect_error=True) - - assert exit_status == 127 - assert result == b'' - assert type(error) == bytes # noqa: E721 - assert b"nonexistent_command" in error - assert b"not found" in error - def test_read__unknown_file(self): """ Test LocalOperations::read with unknown file. diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index a1c6cf44..d92f5e6c 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -11,6 +11,7 @@ import tempfile from ..testgres import InvalidOperationException +from ..testgres import ExecUtilException class TestOsOpsCommon: @@ -28,6 +29,66 @@ def os_ops(self, request: pytest.FixtureRequest) -> OsOperations: assert isinstance(request.param, OsOperations) return request.param + def test_exec_command_success(self, os_ops: OsOperations): + """ + Test exec_command for successful command execution. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + cmd = ["sh", "-c", "python3 --version"] + + response = os_ops.exec_command(cmd) + + assert b'Python 3.' in response + + def test_exec_command_failure(self, os_ops: OsOperations): + """ + Test exec_command for command execution failure. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + cmd = ["sh", "-c", "nonexistent_command"] + + while True: + try: + os_ops.exec_command(cmd) + except ExecUtilException as e: + assert type(e.exit_code) == int # noqa: E721 + assert e.exit_code == 127 + + assert type(e.message) == str # noqa: E721 + assert type(e.error) == bytes # noqa: E721 + + assert e.message.startswith("Utility exited with non-zero code (127). Error:") + assert "nonexistent_command" in e.message + assert "not found" in e.message + assert b"nonexistent_command" in e.error + assert b"not found" in e.error + break + raise Exception("We wait an exception!") + + def test_exec_command_failure__expect_error(self, os_ops: OsOperations): + """ + Test exec_command for command execution failure. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + cmd = ["sh", "-c", "nonexistent_command"] + + exit_status, result, error = os_ops.exec_command(cmd, verbose=True, expect_error=True) + + assert exit_status == 127 + assert result == b'' + assert type(error) == bytes # noqa: E721 + assert b"nonexistent_command" in error + assert b"not found" in error + def test_listdir(self, os_ops: OsOperations): """ Test listdir for listing directory contents. diff --git a/tests/test_remote.py b/tests/test_remote.py index 88c2ede3..a1be7e41 100755 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -20,52 +20,6 @@ def setup(self): ssh_key=os.getenv('RDBMS_TESTPOOL_SSHKEY')) self.operations = RemoteOperations(conn_params) - def test_exec_command_success(self): - """ - Test exec_command for successful command execution. - """ - cmd = "python3 --version" - response = self.operations.exec_command(cmd, wait_exit=True) - - assert b'Python 3.' in response - - def test_exec_command_failure(self): - """ - Test exec_command for command execution failure. - """ - cmd = "nonexistent_command" - while True: - try: - self.operations.exec_command(cmd, verbose=True, wait_exit=True) - except ExecUtilException as e: - assert type(e.exit_code) == int # noqa: E721 - assert e.exit_code == 127 - - assert type(e.message) == str # noqa: E721 - assert type(e.error) == bytes # noqa: E721 - - assert e.message.startswith("Utility exited with non-zero code (127). Error:") - assert "nonexistent_command" in e.message - assert "not found" in e.message - assert b"nonexistent_command" in e.error - assert b"not found" in e.error - break - raise Exception("We wait an exception!") - - def test_exec_command_failure__expect_error(self): - """ - Test exec_command for command execution failure. - """ - cmd = "nonexistent_command" - - exit_status, result, error = self.operations.exec_command(cmd, verbose=True, wait_exit=True, shell=True, expect_error=True) - - assert exit_status == 127 - assert result == b'' - assert type(error) == bytes # noqa: E721 - assert b"nonexistent_command" in error - assert b"not found" in error - def test_is_executable_true(self): """ Test is_executable for an existing executable. From 23cb7616ec23d7050db257913f7f5ee84bbf80cd Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 12:09:09 +0300 Subject: [PATCH 05/11] TestOsOpsCommon is updated (+LocalOperations::is_executable) New tests: - test_is_executable_true - test_is_executable_false - test_makedirs_and_rmdirs_success - test_makedirs_failure LocalOperations::is_executable is corrected --- testgres/operations/local_ops.py | 3 +- tests/test_os_ops_common.py | 60 ++++++++++++++++++++++++++++++++ tests/test_remote.py | 54 ---------------------------- 3 files changed, 62 insertions(+), 55 deletions(-) diff --git a/testgres/operations/local_ops.py b/testgres/operations/local_ops.py index 93a64787..35e94210 100644 --- a/testgres/operations/local_ops.py +++ b/testgres/operations/local_ops.py @@ -156,7 +156,8 @@ def find_executable(self, executable): def is_executable(self, file): # Check if the file is executable - return os.stat(file).st_mode & stat.S_IXUSR + assert stat.S_IXUSR != 0 + return (os.stat(file).st_mode & stat.S_IXUSR) == stat.S_IXUSR def set_env(self, var_name, var_val): # Check if the directory is already in PATH diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index d92f5e6c..b751f4a0 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -89,6 +89,66 @@ def test_exec_command_failure__expect_error(self, os_ops: OsOperations): assert b"nonexistent_command" in error assert b"not found" in error + def test_is_executable_true(self, os_ops: OsOperations): + """ + Test is_executable for an existing executable. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + response = os_ops.is_executable("/bin/sh") + + assert response is True + + def test_is_executable_false(self, os_ops: OsOperations): + """ + Test is_executable for a non-executable. + """ + assert isinstance(os_ops, OsOperations) + + response = os_ops.is_executable(__file__) + + assert response is False + + def test_makedirs_and_rmdirs_success(self, os_ops: OsOperations): + """ + Test makedirs and rmdirs for successful directory creation and removal. + """ + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + cmd = "pwd" + pwd = os_ops.exec_command(cmd, wait_exit=True, encoding='utf-8').strip() + + path = "{}/test_dir".format(pwd) + + # Test makedirs + os_ops.makedirs(path) + assert os.path.exists(path) + assert os_ops.path_exists(path) + + # Test rmdirs + os_ops.rmdirs(path) + assert not os.path.exists(path) + assert not os_ops.path_exists(path) + + def test_makedirs_failure(self, os_ops: OsOperations): + """ + Test makedirs for failure. + """ + # Try to create a directory in a read-only location + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + path = "/root/test_dir" + + # Test makedirs + with pytest.raises(Exception): + os_ops.makedirs(path) + def test_listdir(self, os_ops: OsOperations): """ Test listdir for listing directory contents. diff --git a/tests/test_remote.py b/tests/test_remote.py index a1be7e41..a213a64e 100755 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -6,9 +6,7 @@ from ..testgres import ExecUtilException from ..testgres import RemoteOperations -from ..testgres import LocalOperations from ..testgres import ConnectionParams -from ..testgres import utils as testgres_utils class TestRemoteOperations: @@ -20,58 +18,6 @@ def setup(self): ssh_key=os.getenv('RDBMS_TESTPOOL_SSHKEY')) self.operations = RemoteOperations(conn_params) - def test_is_executable_true(self): - """ - Test is_executable for an existing executable. - """ - local_ops = LocalOperations() - cmd = testgres_utils.get_bin_path2(local_ops, "pg_config") - cmd = local_ops.exec_command([cmd, "--bindir"], encoding="utf-8") - cmd = cmd.rstrip() - cmd = os.path.join(cmd, "pg_config") - response = self.operations.is_executable(cmd) - - assert response is True - - def test_is_executable_false(self): - """ - Test is_executable for a non-executable. - """ - cmd = "python" - response = self.operations.is_executable(cmd) - - assert response is False - - def test_makedirs_and_rmdirs_success(self): - """ - Test makedirs and rmdirs for successful directory creation and removal. - """ - cmd = "pwd" - pwd = self.operations.exec_command(cmd, wait_exit=True, encoding='utf-8').strip() - - path = "{}/test_dir".format(pwd) - - # Test makedirs - self.operations.makedirs(path) - assert os.path.exists(path) - assert self.operations.path_exists(path) - - # Test rmdirs - self.operations.rmdirs(path) - assert not os.path.exists(path) - assert not self.operations.path_exists(path) - - def test_makedirs_failure(self): - """ - Test makedirs for failure. - """ - # Try to create a directory in a read-only location - path = "/root/test_dir" - - # Test makedirs - with pytest.raises(Exception): - self.operations.makedirs(path) - def test_mkdtemp__default(self): path = self.operations.mkdtemp() logging.info("Path is [{0}].".format(path)) From a71b0fb77333763847d0a479b84e40e20432e6e1 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 12:10:51 +0300 Subject: [PATCH 06/11] [FIX] OsOperations::mkstemp is added We forgot about this thing. --- testgres/operations/os_ops.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/testgres/operations/os_ops.py b/testgres/operations/os_ops.py index f20a7a30..3c606871 100644 --- a/testgres/operations/os_ops.py +++ b/testgres/operations/os_ops.py @@ -77,6 +77,9 @@ def pathsep(self): def mkdtemp(self, prefix=None): raise NotImplementedError() + def mkstemp(self, prefix=None): + raise NotImplementedError() + def copytree(self, src, dst): raise NotImplementedError() From a5eabf7a0a8da861bdc635f312c1b670a93b88fd Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 12:12:56 +0300 Subject: [PATCH 07/11] TestOsOpsCommon::test_write_text_file is corrected It did not work in mt-environment. --- tests/test_os_ops_common.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index b751f4a0..b7c5ea96 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -212,7 +212,7 @@ def test_write_text_file(self, os_ops: OsOperations): RunConditions.skip_if_windows() - filename = "/tmp/test_file.txt" + filename = os_ops.mkstemp() data = "Hello, world!" os_ops.write(filename, data, truncate=True) @@ -222,6 +222,8 @@ def test_write_text_file(self, os_ops: OsOperations): assert response == data + data + os_ops.remove_file(filename) + def test_write_binary_file(self, os_ops: OsOperations): """ Test write for writing data to a binary file. From 5111f87095c28f5a249c5ec148032b5ac65f1e25 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 12:31:49 +0300 Subject: [PATCH 08/11] TestOsOpsCommon is updated New tests: - test_mkdtemp__default - test_mkdtemp__custom --- tests/test_local.py | 17 ----------------- tests/test_os_ops_common.py | 21 +++++++++++++++++++++ tests/test_remote.py | 17 ----------------- 3 files changed, 21 insertions(+), 34 deletions(-) diff --git a/tests/test_local.py b/tests/test_local.py index 9514f5f9..7b5e488d 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -3,7 +3,6 @@ import pytest import re -import logging from ..testgres import LocalOperations @@ -14,22 +13,6 @@ class TestLocalOperations: def setup(self): self.operations = LocalOperations() - def test_mkdtemp__default(self): - path = self.operations.mkdtemp() - logging.info("Path is [{0}].".format(path)) - assert os.path.exists(path) - os.rmdir(path) - assert not os.path.exists(path) - - def test_mkdtemp__custom(self): - C_TEMPLATE = "abcdef" - path = self.operations.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) - def test_read__unknown_file(self): """ Test LocalOperations::read with unknown file. diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index b7c5ea96..1a83a080 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -9,6 +9,7 @@ import pytest import re import tempfile +import logging from ..testgres import InvalidOperationException from ..testgres import ExecUtilException @@ -204,6 +205,26 @@ def test_path_exists_false__file(self, os_ops: OsOperations): assert os_ops.path_exists("/etc/nonexistent_path.txt") is False + def test_mkdtemp__default(self, os_ops: OsOperations): + assert isinstance(os_ops, OsOperations) + + 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) + + def test_mkdtemp__custom(self, os_ops: OsOperations): + assert isinstance(os_ops, OsOperations) + + 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) + def test_write_text_file(self, os_ops: OsOperations): """ Test write for writing data to a text file. diff --git a/tests/test_remote.py b/tests/test_remote.py index a213a64e..0d7e8fdc 100755 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -2,7 +2,6 @@ import os import pytest -import logging from ..testgres import ExecUtilException from ..testgres import RemoteOperations @@ -18,22 +17,6 @@ def setup(self): ssh_key=os.getenv('RDBMS_TESTPOOL_SSHKEY')) self.operations = RemoteOperations(conn_params) - def test_mkdtemp__default(self): - path = self.operations.mkdtemp() - logging.info("Path is [{0}].".format(path)) - assert os.path.exists(path) - os.rmdir(path) - assert not os.path.exists(path) - - def test_mkdtemp__custom(self): - C_TEMPLATE = "abcdef" - path = self.operations.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) - def test_rmdirs(self): path = self.operations.mkdtemp() assert os.path.exists(path) From 9f60705c6090defbc24a24240f32b161ab535ec8 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 12:49:28 +0300 Subject: [PATCH 09/11] TestOsOpsCommon is updated New tests: - test_rmdirs - test_rmdirs__01_with_subfolder - test_rmdirs__02_with_file - test_rmdirs__03_with_subfolder_and_file --- tests/test_os_ops_common.py | 67 +++++++++++++++++++++++++++++++++++++ tests/test_remote.py | 59 -------------------------------- 2 files changed, 67 insertions(+), 59 deletions(-) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 1a83a080..3eed5bec 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -225,6 +225,73 @@ def test_mkdtemp__custom(self, os_ops: OsOperations): os.rmdir(path) assert not os.path.exists(path) + def test_rmdirs(self, os_ops: OsOperations): + assert isinstance(os_ops, OsOperations) + + path = os_ops.mkdtemp() + assert os.path.exists(path) + + assert os_ops.rmdirs(path, ignore_errors=False) is True + assert not os.path.exists(path) + + def test_rmdirs__01_with_subfolder(self, os_ops: OsOperations): + assert isinstance(os_ops, OsOperations) + + # folder with subfolder + path = os_ops.mkdtemp() + assert os.path.exists(path) + + dir1 = os.path.join(path, "dir1") + assert not os.path.exists(dir1) + + os_ops.makedirs(dir1) + assert os.path.exists(dir1) + + assert os_ops.rmdirs(path, ignore_errors=False) is True + assert not os.path.exists(path) + assert not os.path.exists(dir1) + + def test_rmdirs__02_with_file(self, os_ops: OsOperations): + assert isinstance(os_ops, OsOperations) + + # folder with file + path = os_ops.mkdtemp() + assert os.path.exists(path) + + file1 = os.path.join(path, "file1.txt") + assert not os.path.exists(file1) + + os_ops.touch(file1) + assert os.path.exists(file1) + + assert os_ops.rmdirs(path, ignore_errors=False) is True + assert not os.path.exists(path) + assert not os.path.exists(file1) + + def test_rmdirs__03_with_subfolder_and_file(self, os_ops: OsOperations): + assert isinstance(os_ops, OsOperations) + + # folder with subfolder and file + path = os_ops.mkdtemp() + assert os.path.exists(path) + + dir1 = os.path.join(path, "dir1") + assert not os.path.exists(dir1) + + os_ops.makedirs(dir1) + assert os.path.exists(dir1) + + file1 = os.path.join(dir1, "file1.txt") + assert not os.path.exists(file1) + + os_ops.touch(file1) + assert os.path.exists(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) + def test_write_text_file(self, os_ops: OsOperations): """ Test write for writing data to a text file. diff --git a/tests/test_remote.py b/tests/test_remote.py index 0d7e8fdc..9f917ba1 100755 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -17,65 +17,6 @@ def setup(self): ssh_key=os.getenv('RDBMS_TESTPOOL_SSHKEY')) self.operations = RemoteOperations(conn_params) - def test_rmdirs(self): - path = self.operations.mkdtemp() - assert os.path.exists(path) - - assert self.operations.rmdirs(path, ignore_errors=False) is True - assert not os.path.exists(path) - - def test_rmdirs__01_with_subfolder(self): - # folder with subfolder - path = self.operations.mkdtemp() - assert os.path.exists(path) - - dir1 = os.path.join(path, "dir1") - assert not os.path.exists(dir1) - - self.operations.makedirs(dir1) - assert os.path.exists(dir1) - - assert self.operations.rmdirs(path, ignore_errors=False) is True - assert not os.path.exists(path) - assert not os.path.exists(dir1) - - def test_rmdirs__02_with_file(self): - # folder with file - path = self.operations.mkdtemp() - assert os.path.exists(path) - - file1 = os.path.join(path, "file1.txt") - assert not os.path.exists(file1) - - self.operations.touch(file1) - assert os.path.exists(file1) - - assert self.operations.rmdirs(path, ignore_errors=False) is True - assert not os.path.exists(path) - assert not os.path.exists(file1) - - def test_rmdirs__03_with_subfolder_and_file(self): - # folder with subfolder and file - path = self.operations.mkdtemp() - assert os.path.exists(path) - - dir1 = os.path.join(path, "dir1") - assert not os.path.exists(dir1) - - self.operations.makedirs(dir1) - assert os.path.exists(dir1) - - file1 = os.path.join(dir1, "file1.txt") - assert not os.path.exists(file1) - - self.operations.touch(file1) - assert os.path.exists(file1) - - assert self.operations.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) - def test_rmdirs__try_to_delete_nonexist_path(self): path = "/root/test_dir" From 7680405879ffe77d56ab15535eb1cab6bf146776 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 12:51:03 +0300 Subject: [PATCH 10/11] TestOsOpsCommon is updated [comments] --- tests/test_os_ops_common.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 3eed5bec..0fbc2f1e 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -403,7 +403,7 @@ def test_read__binary(self, os_ops: OsOperations): def test_read__binary_and_encoding(self, os_ops: OsOperations): """ - Test RemoteOperations::read for binary data and encoding. + Test OsOperations::read for binary data and encoding. """ assert isinstance(os_ops, OsOperations) @@ -416,7 +416,7 @@ def test_read__binary_and_encoding(self, os_ops: OsOperations): def test_read_binary__spec(self, os_ops: OsOperations): """ - Test RemoteOperations::read_binary. + Test OsOperations::read_binary. """ assert isinstance(os_ops, OsOperations) @@ -452,7 +452,7 @@ def test_read_binary__spec(self, os_ops: OsOperations): def test_read_binary__spec__negative_offset(self, os_ops: OsOperations): """ - Test RemoteOperations::read_binary with negative offset. + Test OsOperations::read_binary with negative offset. """ assert isinstance(os_ops, OsOperations) @@ -463,7 +463,7 @@ def test_read_binary__spec__negative_offset(self, os_ops: OsOperations): def test_get_file_size(self, os_ops: OsOperations): """ - Test RemoteOperations::get_file_size. + Test OsOperations::get_file_size. """ assert isinstance(os_ops, OsOperations) From 1716e7c37404780d369c72a895adac3e9391dd6c Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 2 Apr 2025 13:18:54 +0300 Subject: [PATCH 11/11] TestOsOpsCommon::test_touch is added --- tests/test_os_ops_common.py | 16 ++++++++++++++++ tests/test_remote.py | 10 ---------- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 0fbc2f1e..c3944c3b 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -632,3 +632,19 @@ def test_write(self, write_data001: tagWriteData001, os_ops: OsOperations): s = tmp_file.read() assert s == write_data001.result + + def test_touch(self, os_ops: OsOperations): + """ + Test touch for creating a new file or updating access and modification times of an existing file. + """ + assert isinstance(os_ops, OsOperations) + + filename = os_ops.mkstemp() + + # TODO: this test does not check the result of 'touch' command! + + os_ops.touch(filename) + + assert os_ops.isfile(filename) + + os_ops.remove_file(filename) diff --git a/tests/test_remote.py b/tests/test_remote.py index 9f917ba1..565b2d20 100755 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -72,13 +72,3 @@ def test_get_file_size__unk_file(self): assert "Utility exited with non-zero code (1)." in str(x.value) assert "No such file or directory" in str(x.value) assert "/dummy" in str(x.value) - - def test_touch(self): - """ - Test touch for creating a new file or updating access and modification times of an existing file. - """ - filename = "/tmp/test_file.txt" - - self.operations.touch(filename) - - assert self.operations.isfile(filename)