Initial import

Direct copy from testgres 34cc7d1d with minimal changes.

Author: d.kovalenko

.gitignore

@@ -0,0 +1,9 @@
+*.pyc
+*.egg
+*.egg-info/
+.eggs/
+.pytest_cache/
+.vscode/
+__pycache__/
+env/
+venv/

README.md

@@ -0,0 +1,7 @@
+# testgres - os_ops
+
+Subsystem of testgres to work with OS.
+
+## Authors
+
+[Postgres Professional](https://postgrespro.ru/about)

__init__.py

@@ -0,0 +1,6 @@
+[metadata]
+description-file = README.md
+
+[flake8]
+ignore = E501
+exclude = .git,__pycache__,env,venv,testgres/__init__.py

setup.cfg

@@ -0,0 +1,20 @@
+try:
+    from setuptools import setup
+except ImportError:
+    from distutils.core import setup
+
+setup(
+    version="0.0.1",
+    name="testgres.os_ops",
+    packages=[
+        "testgres.operations",
+    ],
+    package_dir={"testgres.operations": "src"},
+    description='Testgres subsystem to work with OS',
+    url='https://github.com/postgrespro/testgres.os_ops',
+    long_description_content_type='text/markdown',
+    license='PostgreSQL',
+    author='Postgres Professional',
+    author_email='testgres@postgrespro.ru',
+    keywords=['testgres'],
+)

setup.py

@@ -0,0 +1,52 @@
+# coding: utf-8
+
+import six
+
+
+class TestgresException(Exception):
+    pass
+
+
+class ExecUtilException(TestgresException):
+    def __init__(self, message=None, command=None, exit_code=0, out=None, error=None):
+        super(ExecUtilException, self).__init__(message)
+
+        self.message = message
+        self.command = command
+        self.exit_code = exit_code
+        self.out = out
+        self.error = error
+
+    def __str__(self):
+        msg = []
+
+        if self.message:
+            msg.append(self.message)
+
+        if self.command:
+            command_s = ' '.join(self.command) if isinstance(self.command, list) else self.command,
+            msg.append(u'Command: {}'.format(command_s))
+
+        if self.exit_code:
+            msg.append(u'Exit code: {}'.format(self.exit_code))
+
+        if self.error:
+            msg.append(u'---- Error:\n{}'.format(self.error))
+
+        if self.out:
+            msg.append(u'---- Out:\n{}'.format(self.out))
+
+        return self.convert_and_join(msg)
+
+    @staticmethod
+    def convert_and_join(msg_list):
+        # Convert each byte element in the list to str
+        str_list = [six.text_type(item, 'utf-8') if isinstance(item, bytes) else six.text_type(item) for item in
+                    msg_list]
+
+        # Join the list into a single string with the specified delimiter
+        return six.text_type('\n').join(str_list)
+
+
+class InvalidOperationException(TestgresException):
+    pass

src/__init__.py

@@ -0,0 +1,55 @@
+import locale
+
+
+class Helpers:
+    @staticmethod
+    def _make_get_default_encoding_func():
+        # locale.getencoding is added in Python 3.11
+        if hasattr(locale, 'getencoding'):
+            return locale.getencoding
+
+        # It must exist
+        return locale.getpreferredencoding
+
+    # Prepared pointer on function to get a name of system codepage
+    _get_default_encoding_func = _make_get_default_encoding_func.__func__()
+
+    @staticmethod
+    def GetDefaultEncoding():
+        #
+        #   Original idea/source was:
+        #
+        #   def os_ops.get_default_encoding():
+        #       if not hasattr(locale, 'getencoding'):
+        #       locale.getencoding = locale.getpreferredencoding
+        #       return locale.getencoding() or 'UTF-8'
+        #
+
+        assert __class__._get_default_encoding_func is not None
+
+        r = __class__._get_default_encoding_func()
+
+        if r:
+            assert r is not None
+            assert type(r) == str  # noqa: E721
+            assert r != ""
+            return r
+
+        # Is it an unexpected situation?
+        return 'UTF-8'
+
+    @staticmethod
+    def PrepareProcessInput(input, encoding):
+        if not input:
+            return None
+
+        if type(input) == str:  # noqa: E721
+            if encoding is None:
+                return input.encode(__class__.GetDefaultEncoding())
+
+            assert type(encoding) == str  # noqa: E721
+            return input.encode(encoding)
+
+        # It is expected!
+        assert type(input) == bytes  # noqa: E721
+        return input