summaryrefslogtreecommitdiff
path: root/test/lib/ansible_test/_internal/io.py
blob: 9d3301a147a2ad0bf25cf710e839368ac9c17e6e (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
"""Functions for disk IO."""
from __future__ import annotations

import errno
import io
import json
import os
import typing 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_obj:
        return file_obj.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,  # type: str
                    content,  # type: t.Any
                    create_directories=False,  # type: bool
                    formatted=True,  # type: bool
                    encoder=None,  # type: t.Optional[t.Callable[[t.Any], t.Any]]
                    ):  # type: (...) -> str
    """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 (',', ':'),
                              cls=encoder,
                              ) + '\n'

    write_text_file(path, text_content, create_directories=create_directories)

    return text_content


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_obj:
        file_obj.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)  # pylint: disable=consider-using-with


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)  # pylint: disable=consider-using-with


class SortedSetEncoder(json.JSONEncoder):
    """Encode sets as sorted lists."""
    def default(self, o):
        """Return a serialized version of the `o` object."""
        if isinstance(o, set):
            return sorted(o)

        return json.JSONEncoder.default(self, o)