mirror of https://github.com/ansible/ansible.git
Code cleanup and refactoring in ansible-test. (#67063)
* Code cleanup in ansible-test. * Split out encoding functions. * Consoldate loading of JSON files. * Split out disk IO functions. * Simplify file access. * Add functions for opening files. * Replace open calls with appropriate functions. * Expose more types from typing module. * Support writing compact JSON. * Add verbosity argument to display.warning. * Add changelog entry. * Update files overlooked during rebase. * Use `io.open` instead of `open`. * Fix file opening for imp.load_module. * Remove use of `r+` mode to access files. * Add missing import. * Fix httptester on Python 2.x. * Clarify changelog fragment. * Consolidate imports. Remove extra newlines. * Fix indirect imports.pull/67089/head
Matt Clay
4 years ago
committed by
GitHub
parent
994a6b0c5a
commit
f4a80bb600
@ -0,0 +1,5 @@
|
||||
minor_changes:
|
||||
- "ansible-test - Support writing compact JSON files instead of formatting and indenting the output."
|
||||
- "ansible-test - Add a verbosity option for displaying warnings."
|
||||
- "ansible-test - Refactor code to consolidate filesystem access and improve handling of encoding."
|
||||
- "ansible-test - General code cleanup."
|
||||
@ -0,0 +1,41 @@
|
||||
"""Functions for encoding and decoding strings."""
|
||||
from __future__ import (absolute_import, division, print_function)
|
||||
__metaclass__ = type
|
||||
|
||||
from . import types as t
|
||||
|
||||
ENCODING = 'utf-8'
|
||||
|
||||
Text = type(u'')
|
||||
|
||||
|
||||
def to_optional_bytes(value, errors='strict'): # type: (t.Optional[t.AnyStr], str) -> t.Optional[bytes]
|
||||
"""Return the given value as bytes encoded using UTF-8 if not already bytes, or None if the value is None."""
|
||||
return None if value is None else to_bytes(value, errors)
|
||||
|
||||
|
||||
def to_optional_text(value, errors='strict'): # type: (t.Optional[t.AnyStr], str) -> t.Optional[t.Text]
|
||||
"""Return the given value as text decoded using UTF-8 if not already text, or None if the value is None."""
|
||||
return None if value is None else to_text(value, errors)
|
||||
|
||||
|
||||
def to_bytes(value, errors='strict'): # type: (t.AnyStr, str) -> bytes
|
||||
"""Return the given value as bytes encoded using UTF-8 if not already bytes."""
|
||||
if isinstance(value, bytes):
|
||||
return value
|
||||
|
||||
if isinstance(value, Text):
|
||||
return value.encode(ENCODING, errors)
|
||||
|
||||
raise Exception('value is not bytes or text: %s' % type(value))
|
||||
|
||||
|
||||
def to_text(value, errors='strict'): # type: (t.AnyStr, str) -> t.Text
|
||||
"""Return the given value as text decoded using UTF-8 if not already text."""
|
||||
if isinstance(value, bytes):
|
||||
return value.decode(ENCODING, errors)
|
||||
|
||||
if isinstance(value, Text):
|
||||
return value
|
||||
|
||||
raise Exception('value is not bytes or text: %s' % type(value))
|
||||
@ -0,0 +1,74 @@
|
||||
"""Functions for disk IO."""
|
||||
from __future__ import (absolute_import, division, print_function)
|
||||
__metaclass__ = type
|
||||
|
||||
import errno
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
|
||||
from . import types as t
|
||||
|
||||
from .encoding import (
|
||||
ENCODING,
|
||||
to_bytes,
|
||||
to_text,
|
||||
)
|
||||
|
||||
|
||||
def read_json_file(path): # type: (t.AnyStr) -> t.Any
|
||||
"""Parse and return the json content from the specified path."""
|
||||
return json.loads(read_text_file(path))
|
||||
|
||||
|
||||
def read_text_file(path): # type: (t.AnyStr) -> t.Text
|
||||
"""Return the contents of the specified path as text."""
|
||||
return to_text(read_binary_file(path))
|
||||
|
||||
|
||||
def read_binary_file(path): # type: (t.AnyStr) -> bytes
|
||||
"""Return the contents of the specified path as bytes."""
|
||||
with open_binary_file(path) as file:
|
||||
return file.read()
|
||||
|
||||
|
||||
def make_dirs(path): # type: (str) -> None
|
||||
"""Create a directory at path, including any necessary parent directories."""
|
||||
try:
|
||||
os.makedirs(to_bytes(path))
|
||||
except OSError as ex:
|
||||
if ex.errno != errno.EEXIST:
|
||||
raise
|
||||
|
||||
|
||||
def write_json_file(path, content, create_directories=False, formatted=True): # type: (str, t.Union[t.List[t.Any], t.Dict[str, t.Any]], bool, bool) -> None
|
||||
"""Write the given json content to the specified path, optionally creating missing directories."""
|
||||
text_content = json.dumps(content, sort_keys=formatted, indent=4 if formatted else None, separators=(', ', ': ') if formatted else (',', ':')) + '\n'
|
||||
write_text_file(path, text_content, create_directories=create_directories)
|
||||
|
||||
|
||||
def write_text_file(path, content, create_directories=False): # type: (str, str, bool) -> None
|
||||
"""Write the given text content to the specified path, optionally creating missing directories."""
|
||||
if create_directories:
|
||||
make_dirs(os.path.dirname(path))
|
||||
|
||||
with open_binary_file(path, 'wb') as file:
|
||||
file.write(to_bytes(content))
|
||||
|
||||
|
||||
def open_text_file(path, mode='r'): # type: (str, str) -> t.TextIO
|
||||
"""Open the given path for text access."""
|
||||
if 'b' in mode:
|
||||
raise Exception('mode cannot include "b" for text files: %s' % mode)
|
||||
|
||||
# noinspection PyTypeChecker
|
||||
return io.open(to_bytes(path), mode, encoding=ENCODING)
|
||||
|
||||
|
||||
def open_binary_file(path, mode='rb'): # type: (str, str) -> t.BinaryIO
|
||||
"""Open the given path for binary access."""
|
||||
if 'b' not in mode:
|
||||
raise Exception('mode must include "b" for binary files: %s' % mode)
|
||||
|
||||
# noinspection PyTypeChecker
|
||||
return io.open(to_bytes(path), mode)
|
||||
Loading…
Reference in New Issue