2104797301

Build Type

github

Committed by GitHub

Commit Message

Merge e928b19ae into f273ef3d3

Pull Request Pull Request #24: Add CMIP file structure, use pyessv controlled vocabularies, and major refactoring

Run Details

230 of 1036 new or added lines in 35 files covered. (22.2%)

13 existing lines in 4 files now uncovered.

705 of 3109 relevant lines covered (22.68%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

28.0

/miranda/archive/ops.py

import logging
import re
import tarfile
import tempfile
import time
from logging import config
from pathlib import Path
from typing import List, Match, Optional, Union

import fabric
from paramiko import SFTPClient, SSHClient, SSHException
from scp import SCPClient, SCPException

from miranda.connect import Connection
from miranda.scripting import LOGGING_CONFIG

logging.config.dictConfig(LOGGING_CONFIG)
__all__ = ["create_archive", "create_remote_directory", "transfer_file", "url_validate"]


def url_validate(target: str) -> Optional[Match[str]]:
    """
    Validates whether a supplied URL is reliably written
    see: https://stackoverflow.com/a/7160778/7322852

    Parameters
    ----------
    target : str

    """
    regex = re.compile(
        r"^(?:http|ftp)s?://"  # http:// or https://
        # domain...
        r"(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|"
        r"localhost|"  # localhost...
        r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})"  # ...or ip
        r"(?::\d+)?"  # optional port
        r"(?:/?|[/?]\S+)$",
        re.IGNORECASE,
    )
    return re.match(regex, target)


def create_remote_directory(
    directory: Union[str, Path],
    transport: Union[SSHClient, fabric.Connection, Connection],
) -> None:
    """
    This calls a "mkdir -p" function to create a folder structure over SFTP/SSH and waits
    for confirmation before continuing

    Parameters
    ----------
    directory : Union[str, Path]
    transport : Union[SSHClient, fabric.Connection, Connection]

    Returns
    -------
    None

    """
    if isinstance(directory, str):
        directory = Path(directory)
    logging.info(f"Creating remote path: {directory}")

    ownership = "0775"
    command = f"mkdir -p -m {ownership} '{directory.as_posix()}'"
    if isinstance(transport, (fabric.Connection, Connection)):
        with transport:
            transport.run(command)
    elif isinstance(transport, (SSHClient, SCPClient)):
        transport.exec_command(command, timeout=1)
        for i in range(5):
            if not directory.exists():
                time.sleep(1)
                continue
            break
    else:
        raise ConnectionError
    return


def create_archive(
    source_files: List[Union[Path, str]],
    destination: Union[Path, str],
    transport: Union[SCPClient, SFTPClient, fabric.Connection, Connection] = None,
    delete: bool = True,
    compression: bool = False,
    recursive: bool = True,
) -> None:
    """

    Parameters
    ----------
    source_files: List[Union[Path, str]]
    destination: Union[Path, str]
    transport: Union[SCPClient, SFTPClient, fabric.Connection, Connection]
    delete: bool
    compression: bool = False
    recursive: bool = True

    Returns
    -------
    None

    """
    if compression:
        write = "w:gz"
    elif not compression:
        write = "w"
    else:
        raise ValueError(f"Compression: {compression}")

    with tempfile.NamedTemporaryFile(delete=delete) as temp:
        archive_file = temp.name
        with tarfile.open(archive_file, write) as tar:
            for name in source_files:
                try:
                    logging.info(f"Tarring {name.name}")
                    tar.add(name.relative_to(Path.cwd()), recursive=recursive)
                except Exception as e:
                    msg = f'File "{name.name}" failed to be tarred: {e}'
                    logging.warning(msg)
            tar.close()
        transfer_file(archive_file, destination, transport)
    return


def transfer_file(
    source_file: Union[Path, str],
    destination_file: Union[Path, str],
    transport: Union[SCPClient, SFTPClient, fabric.Connection, Connection] = None,
) -> bool:
    """

    Parameters
    ----------
    source_file: Union[Path, str]
    destination_file: Union[Path, str]
    transport: Union[SCPClient, SFTPClient, fabric.Connection, Connection]

    Returns
    -------
    bool

    """

    source_file = Path(source_file)
    destination_file = Path(destination_file)

    if transport:
        try:
            logging.info(f"Beginning transfer of {source_file}")
            transport.put(str(source_file), str(destination_file))
            logging.info(
                f"Transferred { Path(destination_file).name} to {Path(destination_file).parent}"
            )

        except (SCPException, SSHException, OSError) as e:
            msg = f'File "{destination_file.name}" failed to be transferred: {e}.'
            logging.warning(msg)
            return False

        logging.info(
            f"Transferred {Path(destination_file).name} to {Path(destination_file).parent}"
        )

    else:
        try:
            destination_file.write_bytes(source_file.read_bytes())
        except (SCPException, SSHException, OSError) as e:
            msg = f'File "{source_file.name}" failed to be copied: {e}'
            logging.error(msg)
            return False
    return True

1	import logging	3✔
2	import re	3✔
3	import tarfile	3✔
4	import tempfile	3✔
5	import time	3✔
6	from logging import config	3✔
7	from pathlib import Path	3✔
8	from typing import List, Match, Optional, Union	3✔
9
10	import fabric	3✔
11	from paramiko import SFTPClient, SSHClient, SSHException	3✔
12	from scp import SCPClient, SCPException	3✔
13
14	from miranda.connect import Connection	3✔
15	from miranda.scripting import LOGGING_CONFIG	3✔
16
17	logging.config.dictConfig(LOGGING_CONFIG)	3✔
18	__all__ = ["create_archive", "create_remote_directory", "transfer_file", "url_validate"]	3✔
19
20
21	def url_validate(target: str) -> Optional[Match[str]]:	3✔
22	"""
23	Validates whether a supplied URL is reliably written
24	see: https://stackoverflow.com/a/7160778/7322852
25
26	Parameters
27	----------
28	target : str
29
30	"""
31	regex = re.compile(	3✔
32	r"^(?:http\|ftp)s?://" # http:// or https://
33	# domain...
34	r"(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?\|[A-Z0-9-]{2,}\.?)\|"
35	r"localhost\|" # localhost...
36	r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})" # ...or ip
37	r"(?::\d+)?" # optional port
38	r"(?:/?\|[/?]\S+)$",
39	re.IGNORECASE,
40	)
41	return re.match(regex, target)	3✔
42
43
44	def create_remote_directory(	3✔
45	directory: Union[str, Path],
46	transport: Union[SSHClient, fabric.Connection, Connection],
47	) -> None:
48	"""
49	This calls a "mkdir -p" function to create a folder structure over SFTP/SSH and waits
50	for confirmation before continuing
51
52	Parameters
53	----------
54	directory : Union[str, Path]
55	transport : Union[SSHClient, fabric.Connection, Connection]
56
57	Returns
58	-------
59	None
60
61	"""
NEW 62	if isinstance(directory, str):	×
NEW 63	directory = Path(directory)	×
NEW 64	logging.info(f"Creating remote path: {directory}")	×
65
66	ownership = "0775"	×
NEW 67	command = f"mkdir -p -m {ownership} '{directory.as_posix()}'"	×
68	if isinstance(transport, (fabric.Connection, Connection)):	×
69	with transport:	×
70	transport.run(command)	×
71	elif isinstance(transport, (SSHClient, SCPClient)):	×
72	transport.exec_command(command, timeout=1)	×
73	for i in range(5):	×
74	if not directory.exists():	×
75	time.sleep(1)	×
76	continue	×
77	break	×
78	else:
79	raise ConnectionError	×
80	return	×
81
82
83	def create_archive(	3✔
84	source_files: List[Union[Path, str]],
85	destination: Union[Path, str],
86	transport: Union[SCPClient, SFTPClient, fabric.Connection, Connection] = None,
87	delete: bool = True,
88	compression: bool = False,
89	recursive: bool = True,
90	) -> None:
91	"""
92
93	Parameters
94	----------
95	source_files: List[Union[Path, str]]
96	destination: Union[Path, str]
97	transport: Union[SCPClient, SFTPClient, fabric.Connection, Connection]
98	delete: bool
99	compression: bool = False
100	recursive: bool = True
101
102	Returns
103	-------
104	None
105
106	"""
107	if compression:	×
108	write = "w:gz"	×
109	elif not compression:	×
110	write = "w"	×
111	else:
NEW 112	raise ValueError(f"Compression: {compression}")	×
113
114	with tempfile.NamedTemporaryFile(delete=delete) as temp:	×
115	archive_file = temp.name	×
116	with tarfile.open(archive_file, write) as tar:	×
117	for name in source_files:	×
118	try:	×
NEW 119	logging.info(f"Tarring {name.name}")	×
120	tar.add(name.relative_to(Path.cwd()), recursive=recursive)	×
121	except Exception as e:	×
NEW 122	msg = f'File "{name.name}" failed to be tarred: {e}'	×
123	logging.warning(msg)	×
124	tar.close()	×
125	transfer_file(archive_file, destination, transport)	×
126	return	×
127
128
129	def transfer_file(	3✔
130	source_file: Union[Path, str],
131	destination_file: Union[Path, str],
132	transport: Union[SCPClient, SFTPClient, fabric.Connection, Connection] = None,
133	) -> bool:
134	"""
135
136	Parameters
137	----------
138	source_file: Union[Path, str]
139	destination_file: Union[Path, str]
140	transport: Union[SCPClient, SFTPClient, fabric.Connection, Connection]
141
142	Returns
143	-------
144	bool
145
146	"""
147
148	source_file = Path(source_file)	×
149	destination_file = Path(destination_file)	×
150
151	if transport:	×
152	try:	×
NEW 153	logging.info(f"Beginning transfer of {source_file}")	×
154	transport.put(str(source_file), str(destination_file))	×
155	logging.info(	×
156	f"Transferred { Path(destination_file).name} to {Path(destination_file).parent}"
157	)
158
NEW 159	except (SCPException, SSHException, OSError) as e:	×
NEW 160	msg = f'File "{destination_file.name}" failed to be transferred: {e}.'	×
161	logging.warning(msg)	×
162	return False	×
163
164	logging.info(	×
165	f"Transferred {Path(destination_file).name} to {Path(destination_file).parent}"
166	)
167
168	else:
169	try:	×
170	destination_file.write_bytes(source_file.read_bytes())	×
NEW 171	except (SCPException, SSHException, OSError) as e:	×
NEW 172	msg = f'File "{source_file.name}" failed to be copied: {e}'	×
173	logging.error(msg)	×
174	return False	×
175	return True	×

Ouranosinc / miranda / 2104797301

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous