Mar 12, 2025 · Mar 12, 2025 · Mar 13, 2025 · Mar 13, 2025 · Mar 13, 2025 · Mar 14, 2025
diff --git a/tests/test_simple.py b/tests/test_simple.py

 from ..testgres import \
    NodeStatus, \
    IsolationLevel, \
    get_new_node

 from ..testgres import \
 # NOTE: those are ugly imports
 from ..testgres import bound_ports
 from ..testgres.utils import PgVer, parse_pg_version
 from ..testgres.utils import file_tail
 from ..testgres.node import ProcessProxy


                r.execute('select 1')
                r.safe_psql('select 1')

    def test_auto_name(self):
        with get_new_node().init(allow_streaming=True).start() as m:
            with m.replicate().start() as r:
                # check that nodes are running
                assert (m.status())
                assert (r.status())

                # check their names
                assert (m.name != r.name)
                assert ('testgres' in m.name)
                assert ('testgres' in r.name)

    def test_file_tail(self):
        s1 = "the quick brown fox jumped over that lazy dog\n"
        s2 = "abc\n"
        s3 = "def\n"

        with tempfile.NamedTemporaryFile(mode='r+', delete=True) as f:
            sz = 0
            while sz < 3 * 8192:
                sz += len(s1)
                f.write(s1)
            f.write(s2)
            f.write(s3)

            f.seek(0)
            lines = file_tail(f, 3)
            assert (lines[0] == s1)
            assert (lines[1] == s2)
            assert (lines[2] == s3)

            f.seek(0)
            lines = file_tail(f, 1)
            assert (lines[0] == s3)

    def test_isolation_levels(self):
        with get_new_node().init().start() as node:
            with node.connect() as con:
                # string levels
                con.begin('Read Uncommitted').commit()
                con.begin('Read Committed').commit()
                con.begin('Repeatable Read').commit()
                con.begin('Serializable').commit()

                # enum levels
                con.begin(IsolationLevel.ReadUncommitted).commit()
                con.begin(IsolationLevel.ReadCommitted).commit()
                con.begin(IsolationLevel.RepeatableRead).commit()
                con.begin(IsolationLevel.Serializable).commit()

                # check wrong level
                with pytest.raises(expected_exception=QueryException):
                    con.begin('Garbage').commit()

    def test_ports_management(self):
        assert bound_ports is not None
        assert type(bound_ports) == set  # noqa: E721
diff --git a/tests/test_simple_remote.py b/tests/test_simple_remote.py
    pop_config, testgres_config

 from ..testgres import \
    NodeStatus, \
    IsolationLevel
    NodeStatus

 from ..testgres import \
    get_bin_path, \
 # NOTE: those are ugly imports
 from ..testgres import bound_ports
 from ..testgres.utils import PgVer
 from ..testgres.utils import file_tail
 from ..testgres.node import ProcessProxy


                assert (res_exec == [(1,)])
                assert (res_psql == b'1\n')

    def test_auto_name(self):
        with __class__.helper__get_node().init(allow_streaming=True).start() as m:
            with m.replicate().start() as r:
                # check that nodes are running
                assert (m.status())
                assert (r.status())

                # check their names
                assert (m.name != r.name)
                assert ('testgres' in m.name)
                assert ('testgres' in r.name)

    def test_file_tail(self):
        s1 = "the quick brown fox jumped over that lazy dog\n"
        s2 = "abc\n"
        s3 = "def\n"

        with tempfile.NamedTemporaryFile(mode='r+', delete=True) as f:
            sz = 0
            while sz < 3 * 8192:
                sz += len(s1)
                f.write(s1)
            f.write(s2)
            f.write(s3)

            f.seek(0)
            lines = file_tail(f, 3)
            assert (lines[0] == s1)
            assert (lines[1] == s2)
            assert (lines[2] == s3)

            f.seek(0)
            lines = file_tail(f, 1)
            assert (lines[0] == s3)

    def test_isolation_levels(self):
        with __class__.helper__get_node().init().start() as node:
            with node.connect() as con:
                # string levels
                con.begin('Read Uncommitted').commit()
                con.begin('Read Committed').commit()
                con.begin('Repeatable Read').commit()
                con.begin('Serializable').commit()

                # enum levels
                con.begin(IsolationLevel.ReadUncommitted).commit()
                con.begin(IsolationLevel.ReadCommitted).commit()
                con.begin(IsolationLevel.RepeatableRead).commit()
                con.begin(IsolationLevel.Serializable).commit()

                # check wrong level
                with pytest.raises(expected_exception=QueryException):
                    con.begin('Garbage').commit()

    def test_ports_management(self):
        assert bound_ports is not None
        assert type(bound_ports) == set  # noqa: E721
diff --git a/tests/test_testgres_common.py b/tests/test_testgres_common.py
 from ..testgres.node import PgVer
 from ..testgres.node import PostgresNode
 from ..testgres.utils import get_pg_version2
 from ..testgres.utils import file_tail
 from ..testgres import ProcessType
 from ..testgres import IsolationLevel
 from ..testgres import TestgresException
 from ..testgres import StartNodeException
 from ..testgres import QueryException
 import six
 import logging
 import time
 import tempfile


 class TestTestgresCommon:
        str(ExecUtilException('msg', 'cmd', 1, 'out'))
        str(QueryException('msg', 'query'))

    def test_auto_name(self, os_ops: OsOperations):
        assert isinstance(os_ops, OsOperations)

        with __class__.helper__get_node(os_ops).init(allow_streaming=True).start() as m:
            with m.replicate().start() as r:
                # check that nodes are running
                assert (m.status())
                assert (r.status())

                # check their names
                assert (m.name != r.name)
                assert ('testgres' in m.name)
                assert ('testgres' in r.name)

    def test_file_tail(self):
        s1 = "the quick brown fox jumped over that lazy dog\n"
        s2 = "abc\n"
        s3 = "def\n"

        with tempfile.NamedTemporaryFile(mode='r+', delete=True) as f:
            sz = 0
            while sz < 3 * 8192:
                sz += len(s1)
                f.write(s1)
            f.write(s2)
            f.write(s3)

            f.seek(0)
            lines = file_tail(f, 3)
            assert (lines[0] == s1)
            assert (lines[1] == s2)
            assert (lines[2] == s3)

            f.seek(0)
            lines = file_tail(f, 1)
            assert (lines[0] == s3)

    def test_isolation_levels(self, os_ops: OsOperations):
        assert isinstance(os_ops, OsOperations)
        with __class__.helper__get_node(os_ops).init().start() as node:
            with node.connect() as con:
                # string levels
                con.begin('Read Uncommitted').commit()
                con.begin('Read Committed').commit()
                con.begin('Repeatable Read').commit()
                con.begin('Serializable').commit()

                # enum levels
                con.begin(IsolationLevel.ReadUncommitted).commit()
                con.begin(IsolationLevel.ReadCommitted).commit()
                con.begin(IsolationLevel.RepeatableRead).commit()
                con.begin(IsolationLevel.Serializable).commit()

                # check wrong level
                with pytest.raises(expected_exception=QueryException):
                    con.begin('Garbage').commit()

    @staticmethod
    def helper__get_node(os_ops: OsOperations, name=None):
        assert isinstance(os_ops, OsOperations)
Original file line number	Diff line number	Diff line change
Expand Up		@@ -34,7 +34,6 @@

		from ..testgres import \
		NodeStatus, \
		IsolationLevel, \
		get_new_node

		from ..testgres import \
Expand All		@@ -49,7 +48,6 @@
		# NOTE: those are ugly imports
		from ..testgres import bound_ports
		from ..testgres.utils import PgVer, parse_pg_version
		from ..testgres.utils import file_tail
		from ..testgres.node import ProcessProxy


Expand DownExpand Up		@@ -953,60 +951,6 @@ def test_unix_sockets(self):
		r.execute('select 1')
		r.safe_psql('select 1')

		def test_auto_name(self):
		with get_new_node().init(allow_streaming=True).start() as m:
		with m.replicate().start() as r:
		# check that nodes are running
		assert (m.status())
		assert (r.status())

		# check their names
		assert (m.name != r.name)
		assert ('testgres' in m.name)
		assert ('testgres' in r.name)

		def test_file_tail(self):
		s1 = "the quick brown fox jumped over that lazy dog\n"
		s2 = "abc\n"
		s3 = "def\n"

		with tempfile.NamedTemporaryFile(mode='r+', delete=True) as f:
		sz = 0
		while sz < 3 * 8192:
		sz += len(s1)
		f.write(s1)
		f.write(s2)
		f.write(s3)

		f.seek(0)
		lines = file_tail(f, 3)
		assert (lines[0] == s1)
		assert (lines[1] == s2)
		assert (lines[2] == s3)

		f.seek(0)
		lines = file_tail(f, 1)
		assert (lines[0] == s3)

		def test_isolation_levels(self):
		with get_new_node().init().start() as node:
		with node.connect() as con:
		# string levels
		con.begin('Read Uncommitted').commit()
		con.begin('Read Committed').commit()
		con.begin('Repeatable Read').commit()
		con.begin('Serializable').commit()

		# enum levels
		con.begin(IsolationLevel.ReadUncommitted).commit()
		con.begin(IsolationLevel.ReadCommitted).commit()
		con.begin(IsolationLevel.RepeatableRead).commit()
		con.begin(IsolationLevel.Serializable).commit()

		# check wrong level
		with pytest.raises(expected_exception=QueryException):
		con.begin('Garbage').commit()

		def test_ports_management(self):
		assert bound_ports is not None
		assert type(bound_ports) == set # noqa: E721
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -34,8 +34,7 @@
		pop_config, testgres_config

		from ..testgres import \
		NodeStatus, \
		IsolationLevel
		NodeStatus

		from ..testgres import \
		get_bin_path, \
Expand All		@@ -49,7 +48,6 @@
		# NOTE: those are ugly imports
		from ..testgres import bound_ports
		from ..testgres.utils import PgVer
		from ..testgres.utils import file_tail
		from ..testgres.node import ProcessProxy


Expand DownExpand Up		@@ -1025,60 +1023,6 @@ def test_unix_sockets(self):
		assert (res_exec == [(1,)])
		assert (res_psql == b'1\n')

		def test_auto_name(self):
		with __class__.helper__get_node().init(allow_streaming=True).start() as m:
		with m.replicate().start() as r:
		# check that nodes are running
		assert (m.status())
		assert (r.status())

		# check their names
		assert (m.name != r.name)
		assert ('testgres' in m.name)
		assert ('testgres' in r.name)

		def test_file_tail(self):
		s1 = "the quick brown fox jumped over that lazy dog\n"
		s2 = "abc\n"
		s3 = "def\n"

		with tempfile.NamedTemporaryFile(mode='r+', delete=True) as f:
		sz = 0
		while sz < 3 * 8192:
		sz += len(s1)
		f.write(s1)
		f.write(s2)
		f.write(s3)

		f.seek(0)
		lines = file_tail(f, 3)
		assert (lines[0] == s1)
		assert (lines[1] == s2)
		assert (lines[2] == s3)

		f.seek(0)
		lines = file_tail(f, 1)
		assert (lines[0] == s3)

		def test_isolation_levels(self):
		with __class__.helper__get_node().init().start() as node:
		with node.connect() as con:
		# string levels
		con.begin('Read Uncommitted').commit()
		con.begin('Read Committed').commit()
		con.begin('Repeatable Read').commit()
		con.begin('Serializable').commit()

		# enum levels
		con.begin(IsolationLevel.ReadUncommitted).commit()
		con.begin(IsolationLevel.ReadCommitted).commit()
		con.begin(IsolationLevel.RepeatableRead).commit()
		con.begin(IsolationLevel.Serializable).commit()

		# check wrong level
		with pytest.raises(expected_exception=QueryException):
		con.begin('Garbage').commit()

		def test_ports_management(self):
		assert bound_ports is not None
		assert type(bound_ports) == set # noqa: E721
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -5,7 +5,9 @@
		from ..testgres.node import PgVer
		from ..testgres.node import PostgresNode
		from ..testgres.utils import get_pg_version2
		from ..testgres.utils import file_tail
		from ..testgres import ProcessType
		from ..testgres import IsolationLevel
		from ..testgres import TestgresException
		from ..testgres import StartNodeException
		from ..testgres import QueryException
Expand All		@@ -15,6 +17,7 @@
		import six
		import logging
		import time
		import tempfile


		class TestTestgresCommon:
Expand DownExpand Up		@@ -186,6 +189,63 @@ def test_exceptions(self):
		str(ExecUtilException('msg', 'cmd', 1, 'out'))
		str(QueryException('msg', 'query'))

		def test_auto_name(self, os_ops: OsOperations):
		assert isinstance(os_ops, OsOperations)

		with __class__.helper__get_node(os_ops).init(allow_streaming=True).start() as m:
		with m.replicate().start() as r:
		# check that nodes are running
		assert (m.status())
		assert (r.status())

		# check their names
		assert (m.name != r.name)
		assert ('testgres' in m.name)
		assert ('testgres' in r.name)

		def test_file_tail(self):
		s1 = "the quick brown fox jumped over that lazy dog\n"
		s2 = "abc\n"
		s3 = "def\n"

		with tempfile.NamedTemporaryFile(mode='r+', delete=True) as f:
		sz = 0
		while sz < 3 * 8192:
		sz += len(s1)
		f.write(s1)
		f.write(s2)
		f.write(s3)

		f.seek(0)
		lines = file_tail(f, 3)
		assert (lines[0] == s1)
		assert (lines[1] == s2)
		assert (lines[2] == s3)

		f.seek(0)
		lines = file_tail(f, 1)
		assert (lines[0] == s3)

		def test_isolation_levels(self, os_ops: OsOperations):
		assert isinstance(os_ops, OsOperations)
		with __class__.helper__get_node(os_ops).init().start() as node:
		with node.connect() as con:
		# string levels
		con.begin('Read Uncommitted').commit()
		con.begin('Read Committed').commit()
		con.begin('Repeatable Read').commit()
		con.begin('Serializable').commit()

		# enum levels
		con.begin(IsolationLevel.ReadUncommitted).commit()
		con.begin(IsolationLevel.ReadCommitted).commit()
		con.begin(IsolationLevel.RepeatableRead).commit()
		con.begin(IsolationLevel.Serializable).commit()

		# check wrong level
		with pytest.raises(expected_exception=QueryException):
		con.begin('Garbage').commit()

		@staticmethod
		def helper__get_node(os_ops: OsOperations, name=None):
		assert isinstance(os_ops, OsOperations)
Expand Down