<commit_before># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

"""
Tests for ``flocker.cli._sshconfig``.
"""

from operator import setitem, delitem
from os import devnull, environ, kill
from os.path import expanduser
from signal import SIGKILL
from socket import socket
from subprocess import check_output, check_call

from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
from twisted.conch.ssh.keys import Key
from twisted.internet.threads import deferToThread

from .. import configure_ssh
from .._sshconfig import OpenSSHConfiguration
from ...testtools import create_ssh_server


def goodlines(path):
    """
    Return a list of lines read from ``path`` excluding those that are blank
    or begin with ``#``.

    :param FilePath path: The path to the file to read.

    :return: A ``list`` of ``bytes`` giving good lines from the file.
    """
    return list(line for line in path.getContent().splitlines()
                if line and not line.strip().startswith(b"#"))


class ConfigureSSHTests(TestCase):
    """
    Tests for ``configure_ssh``.
    """
    def setUp(self):
        self.ssh_config = FilePath(self.mktemp())
        self.server = create_ssh_server(self.ssh_config)
        self.addCleanup(self.server.restore)
        self.flocker_config = FilePath(self.mktemp())
        self.config = OpenSSHConfiguration(
            ssh_config_path=self.ssh_config,
            flocker_path=self.flocker_config)
        self.configure_ssh = self.config.configure_ssh

        output = check_output([b"ssh-agent", b"-c"]).splitlines()

        # ``configure_ssh`` expects ``ssh`` to already be able to
        # authenticate against the server.  Set up an ssh-agent to
        # help it do that against our testing server.

        # setenv SSH_AUTH_SOCK /tmp/ssh-5EfGti8RPQbQ/agent.6390;
        # setenv SSH_AGENT_PID 6391;
        # echo Agent pid 6391;
        sock = output[0].split()[2][:-1]
        pid = output[1].split()[2][:-1]

        self.addCleanup(lambda: kill(int(pid), SIGKILL))

        def patchdict(k, v):
            if k in environ:
                self.addCleanup(lambda old=environ[k]: setitem(environ, k, old))
            else:
                self.addCleanup(lambda: delitem(environ, k))

            environ[k] = v

        patchdict(b"SSH_AUTH_SOCK", sock)
        patchdict(b"SSH_AGENT_PID", pid)

        with open(devnull, "w") as discard:
            # See https://github.com/clusterhq/flocker/issues/192
            check_call(
                [b"ssh-add", self.server.key_path.path],
                stdout=discard, stderr=discard)

    def test_connection_failed(self):
        """<commit_msg>Fix too-long line.<commit_after># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

"""
Tests for ``flocker.cli._sshconfig``.
"""

from operator import setitem, delitem
from os import devnull, environ, kill
from os.path import expanduser
from signal import SIGKILL
from socket import socket
from subprocess import check_output, check_call

from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
from twisted.conch.ssh.keys import Key
from twisted.internet.threads import deferToThread

from .. import configure_ssh
from .._sshconfig import OpenSSHConfiguration
from ...testtools import create_ssh_server


def goodlines(path):
    """
    Return a list of lines read from ``path`` excluding those that are blank
    or begin with ``#``.

    :param FilePath path: The path to the file to read.

    :return: A ``list`` of ``bytes`` giving good lines from the file.
    """
    return list(line for line in path.getContent().splitlines()
                if line and not line.strip().startswith(b"#"))


class ConfigureSSHTests(TestCase):
    """
    Tests for ``configure_ssh``.
    """
    def setUp(self):
        self.ssh_config = FilePath(self.mktemp())
        self.server = create_ssh_server(self.ssh_config)
        self.addCleanup(self.server.restore)
        self.flocker_config = FilePath(self.mktemp())
        self.config = OpenSSHConfiguration(
            ssh_config_path=self.ssh_config,
            flocker_path=self.flocker_config)
        self.configure_ssh = self.config.configure_ssh

        output = check_output([b"ssh-agent", b"-c"]).splitlines()

        # ``configure_ssh`` expects ``ssh`` to already be able to
        # authenticate against the server.  Set up an ssh-agent to
        # help it do that against our testing server.

        # setenv SSH_AUTH_SOCK /tmp/ssh-5EfGti8RPQbQ/agent.6390;
        # setenv SSH_AGENT_PID 6391;
        # echo Agent pid 6391;
        sock = output[0].split()[2][:-1]
        pid = output[1].split()[2][:-1]

        self.addCleanup(lambda: kill(int(pid), SIGKILL))

        def patchdict(k, v):
            if k in environ:
                self.addCleanup(
                    lambda old=environ[k]: setitem(environ, k, old))
            else:
                self.addCleanup(lambda: delitem(environ, k))

            environ[k] = v

        patchdict(b"SSH_AUTH_SOCK", sock)
        patchdict(b"SSH_AGENT_PID", pid)

        with open(devnull, "w") as discard:
            # See https://github.com/clusterhq/flocker/issues/192
            check_call(
                [b"ssh-add", self.server.key_path.path],
                stdout=discard, stderr=discard)

    def test_connection_failed(self):
        """
{'fraction_non_alphanumeric': 0.10761834319526627, 'fraction_numerical': 0.008505917159763314, 'mean_word_length': 2.6997264021887823, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1070089', 'n_tokens_mistral': 1724, 'n_tokens_neox': 1606, 'n_words': 441}