Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_remote(self, mock_ping):
p = Client(
uri='unix:/run/podman',
interface='io.podman',
remote_uri='ssh://user@hostname/run/podman/podman',
identity_file='~/.ssh/id_rsa')
self.assertIsInstance(p._client, BaseClient)
mock_ping.assert_called_once_with()
def test_local(self, mock_ping):
p = Client(
uri='unix:/run/podman',
interface='io.podman',
)
self.assertIsInstance(p._client, LocalClient)
self.assertIsInstance(p._client, BaseClient)
mock_ping.assert_called_once_with()
return RemoteClient(
Context(
uri,
interface,
local_path,
remote.path,
remote.username,
remote.hostname,
remote.port,
kwargs.get('identity_file'),
kwargs.get('ignore_hosts'),
kwargs.get('known_hosts'),
))
class LocalClient(BaseClient):
"""Context manager for API workers to access varlink."""
def __enter__(self):
"""Enter context for LocalClient."""
self._client = VarlinkClient(address=self._context.uri)
self._iface = self._client.open(self._context.interface)
return self._iface
def __exit__(self, e_type, e, e_traceback):
"""Cleanup context for LocalClient."""
if hasattr(self._client, 'close'):
self._client.close()
self._iface.close()
if isinstance(e, VarlinkError):
raise error_factory(e)
"""Enter context for LocalClient."""
self._client = VarlinkClient(address=self._context.uri)
self._iface = self._client.open(self._context.interface)
return self._iface
def __exit__(self, e_type, e, e_traceback):
"""Cleanup context for LocalClient."""
if hasattr(self._client, 'close'):
self._client.close()
self._iface.close()
if isinstance(e, VarlinkError):
raise error_factory(e)
class RemoteClient(BaseClient):
"""Context manager for API workers to access remote varlink."""
def __init__(self, context):
"""Construct RemoteCLient."""
super().__init__(context)
self._portal = Portal()
def __enter__(self):
"""Context manager for API workers to access varlink."""
tunnel = self._portal.get(self._context.uri)
if tunnel is None:
tunnel = Tunnel(self._context).bore()
self._portal[self._context.uri] = tunnel
try:
self._client = VarlinkClient(address=self._context.uri)