Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

OsOps::read methods were corrected (text mode)#171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Merged
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 29 additions & 6 deletionstestgres/operations/local_ops.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -10,6 +10,7 @@
import psutil

from ..exceptions import ExecUtilException
from ..exceptions import InvalidOperationException
from .os_ops import ConnectionParams, OsOperations, pglib, get_default_encoding
from .raise_error import RaiseError
from .helpers import Helpers
Expand DownExpand Up@@ -266,13 +267,35 @@ def touch(self, filename):
os.utime(filename, None)

def read(self, filename, encoding=None, binary=False):
mode = "rb" if binary else "r"
with open(filename, mode) as file:
assert type(filename) == str # noqa: E721
assert encoding is None or type(encoding) == str # noqa: E721
assert type(binary) == bool # noqa: E721

if binary:
if encoding is not None:
raise InvalidOperationException("Enconding is not allowed for read binary operation")

return self._read__binary(filename)

# python behavior
assert (None or "abc") == "abc"
assert ("" or "abc") == "abc"

return self._read__text_with_encoding(filename, encoding or get_default_encoding())

def _read__text_with_encoding(self, filename, encoding):
assert type(filename) == str # noqa: E721
assert type(encoding) == str # noqa: E721
with open(filename, mode='r', encoding=encoding) as file: # open in a text mode
content = file.read()
assert type(content) == str # noqa: E721
return content

def _read__binary(self, filename):
assert type(filename) == str # noqa: E721
with open(filename, 'rb') as file: # open in a binary mode
content = file.read()
if binary:
return content
if isinstance(content, bytes):
return content.decode(encoding or get_default_encoding())
assert type(content) == bytes # noqa: E721
return content

def readlines(self, filename, num_lines=0, binary=False, encoding=None):
Expand Down
38 changes: 33 additions & 5 deletionstestgres/operations/remote_ops.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -3,6 +3,7 @@
import platform
import subprocess
import tempfile
import io

# we support both pg8000 and psycopg2
try:
Expand All@@ -14,6 +15,7 @@
raise ImportError("You must have psycopg2 or pg8000 modules installed")

from ..exceptions import ExecUtilException
from ..exceptions import InvalidOperationException
from .os_ops import OsOperations, ConnectionParams, get_default_encoding
from .raise_error import RaiseError
from .helpers import Helpers
Expand DownExpand Up@@ -319,13 +321,39 @@ def touch(self, filename):
self.exec_command("touch {}".format(filename))

def read(self, filename, binary=False, encoding=None):
cmd = "cat {}".format(filename)
result = self.exec_command(cmd, encoding=encoding)
assert type(filename) == str # noqa: E721
assert encoding is None or type(encoding) == str # noqa: E721
assert type(binary) == bool # noqa: E721

if not binary and result:
result = result.decode(encoding or get_default_encoding())
if binary:
if encoding is not None:
raise InvalidOperationException("Enconding is not allowed for read binary operation")

return result
return self._read__binary(filename)

# python behavior
assert (None or "abc") == "abc"
assert ("" or "abc") == "abc"

return self._read__text_with_encoding(filename, encoding or get_default_encoding())

def _read__text_with_encoding(self, filename, encoding):
assert type(filename) == str # noqa: E721
assert type(encoding) == str # noqa: E721
content = self._read__binary(filename)
assert type(content) == bytes # noqa: E721
buf0 = io.BytesIO(content)
buf1 = io.TextIOWrapper(buf0, encoding=encoding)
content_s = buf1.read()
assert type(content_s) == str # noqa: E721
return content_s

def _read__binary(self, filename):
assert type(filename) == str # noqa: E721
cmd = ["cat", filename]
content = self.exec_command(cmd)
assert type(content) == bytes # noqa: E721
return content

def readlines(self, filename, num_lines=0, binary=False, encoding=None):
if num_lines > 0:
Expand Down
66 changes: 65 additions & 1 deletiontests/test_local.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
import re

from testgres import ExecUtilException
from testgres import InvalidOperationException
from testgres import LocalOperations

from .helpers.run_conditions import RunConditions
Expand DownExpand Up@@ -56,6 +57,67 @@ def test_exec_command_failure__expect_error(self):
assert exit_status == 127
assert result == b''

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.
"""

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.
Expand DownExpand Up@@ -95,7 +157,9 @@ def test_read_binary__spec__unk_file(self):
Test LocalOperations::read_binary with unknown file.
"""

with pytest.raises(FileNotFoundError, match=re.escape("[Errno 2] No such file or directory: '/dummy'")):
with pytest.raises(
FileNotFoundError,
match=re.escape("[Errno 2] No such file or directory: '/dummy'")):
self.operations.read_binary("/dummy", 0)

def test_get_file_size(self):
Expand Down
64 changes: 64 additions & 0 deletionstests/test_remote.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
import re

from testgres import ExecUtilException
from testgres import InvalidOperationException
from testgres import RemoteOperations
from testgres import ConnectionParams

Expand DownExpand Up@@ -182,6 +183,69 @@ 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.
"""

with pytest.raises(
ExecUtilException,
match=re.escape("cat: /dummy: No such file or directory")):
self.operations.read("/dummy")

def test_read_binary__spec(self):
"""
Test RemoteOperations::read_binary.
Expand Down
[8]ページ先頭
©2009-2025 Movatter.jp