2157378933

Build Type

github

Committed by GitHub

Commit Message

Merge c79a04131 into 9ac032fc5

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

Run Details

241 of 1107 new or added lines in 35 files covered. (21.77%)

13 existing lines in 4 files now uncovered.

735 of 3247 relevant lines covered (22.64%)

0.68 hits per line

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

84.44

/miranda/connect.py

import logging.config
from getpass import getpass
from pathlib import Path
from typing import Union

import fabric
from paramiko import SSHClient
from scp import SCPClient

from .scripting import LOGGING_CONFIG

logging.config.dictConfig(LOGGING_CONFIG)

__all__ = ["Connection"]


class Connection:
    def __init__(
        self,
        username: Union[str, Path] = None,
        host: Union[str, Path] = None,
        protocol: str = "sftp",
        *args,
        **kwargs,
    ):
        self.user = username or input("Enter username: ")
        self.host = host or input("Enter host URL: ")
        self._args = list(*args)
        self._kwargs = {**kwargs}
        self.__c = None

        if protocol.lower() in ["sftp", "scp"]:
            self.protocol = protocol.lower()
        else:
            raise ValueError('Protocol must be "sftp" or "scp".')

    def update(self, **kwargs):
        self._kwargs = kwargs

    def __call__(self, **kwargs):
        self.update(**kwargs)
        return self

    def __str__(self):
        return f"Connection to {self.host} as {self.user}"

    def __repr__(self):
        return "<{}.{} object at {}>".format(
            self.__class__.__module__, self.__class__.__name__, hex(id(self))
        )

    def connect(self, **kwargs):
        try:
            keywords = (
                dict(**kwargs)
                or dict(**self._kwargs)
                or dict(password=getpass("Enter password: "))
            )
            if self.protocol == "sftp":
                c = fabric.Connection(
                    host=self.host, user=self.user, connect_kwargs=keywords
                )
                self.__c = c
            else:
                c = SSHClient()
                c.connect(
                    self.host,
                    username=self.user,
                    password=self._kwargs["password"] or getpass("Enter password: "),
                )
                self.__c = SCPClient(c.get_transport())

            return self.__c
        except Exception as e:
            raise e

    def __enter__(self, **kwargs):
        return self.connect()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.__c.close()

1	import logging.config	3✔
2	from getpass import getpass	3✔
3	from pathlib import Path	3✔
4	from typing import Union	3✔
5
6	import fabric	3✔
7	from paramiko import SSHClient	3✔
8	from scp import SCPClient	3✔
9
10	from .scripting import LOGGING_CONFIG	3✔
11
12	logging.config.dictConfig(LOGGING_CONFIG)	3✔
13
14	__all__ = ["Connection"]	3✔
15
16
17	class Connection:	3✔
18	def __init__(	3✔
19	self,
20	username: Union[str, Path] = None,
21	host: Union[str, Path] = None,
22	protocol: str = "sftp",
23	*args,
24	**kwargs,
25	):
26	self.user = username or input("Enter username: ")	3✔
27	self.host = host or input("Enter host URL: ")	3✔
28	self._args = list(*args)	3✔
29	self._kwargs = {**kwargs}	3✔
30	self.__c = None	3✔
31
32	if protocol.lower() in ["sftp", "scp"]:	3✔
33	self.protocol = protocol.lower()	3✔
34	else:
35	raise ValueError('Protocol must be "sftp" or "scp".')	×
36
37	def update(self, **kwargs):	3✔
38	self._kwargs = kwargs	3✔
39
40	def __call__(self, **kwargs):	3✔
41	self.update(**kwargs)	3✔
42	return self	3✔
43
44	def __str__(self):	3✔
NEW 45	return f"Connection to {self.host} as {self.user}"	×
46
47	def __repr__(self):	3✔
48	return "<{}.{} object at {}>".format(	3✔
49	self.__class__.__module__, self.__class__.__name__, hex(id(self))
50	)
51
52	def connect(self, **kwargs):	3✔
53	try:	3✔
54	keywords = (	3✔
55	dict(**kwargs)
56	or dict(**self._kwargs)
57	or dict(password=getpass("Enter password: "))
58	)
59	if self.protocol == "sftp":	3✔
60	c = fabric.Connection(	3✔
61	host=self.host, user=self.user, connect_kwargs=keywords
62	)
63	self.__c = c	3✔
64	else:
65	c = SSHClient()	×
66	c.connect(	×
67	self.host,
68	username=self.user,
69	password=self._kwargs["password"] or getpass("Enter password: "),
70	)
71	self.__c = SCPClient(c.get_transport())	×
72
73	return self.__c	3✔
74	except Exception as e:	×
75	raise e	×
76
77	def __enter__(self, **kwargs):	3✔
78	return self.connect()	3✔
79
80	def __exit__(self, exc_type, exc_val, exc_tb):	3✔
81	self.__c.close()	3✔

Ouranosinc / miranda / 2157378933

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