📦 wagoodman / bridgy

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
96
97
98
99
100
101
102
103
104
105import pytest
import shlex
import re
import os

from bridgy.command import Ssh
from bridgy.inventory import InventorySet, Instance, inventory
from bridgy.inventory.aws import AwsInventory
from bridgy.error import BadInstanceError, BadConfigError, MissingBastionHost
from bridgy.config import Config

instance = Instance('name', 'address.com')

whitespace_pattern = re.compile(r'\W+')

def assert_command_results(result1, result2):
    result1 = shlex.split(result1)
    result2 = shlex.split(result2)

    assert len(result1) == len(result2)

    for item1, item2 in zip(result1, result2):
        item1 = whitespace_pattern.sub(' ', item1)
        item2 = whitespace_pattern.sub(' ', item2)
        assert item1 == item2

def test_ssh_command_go_case():
    config = Config({
        'ssh': {}
    })
    sshObj = Ssh(config, instance)
    assert_command_results(sshObj.command, 'ssh -t address.com')

def test_ssh_command_go_case_no_options():
    config = Config({})
    sshObj = Ssh(config, instance)
    assert_command_results(sshObj.command, 'ssh -t address.com')

def test_ssh_command_user():
    config = Config({
        'ssh': {
            'user': 'username'
        }
    })
    sshObj = Ssh(config, instance)
    assert_command_results(sshObj.command, 'ssh -t username@address.com')

def test_ssh_command_options():
    config = Config({
        'ssh': {
            'user': 'username',
            'options': '-C -o ServerAliveInterval=255'
        }
    })
    sshObj = Ssh(config, instance)
    assert_command_results(sshObj.command, 'ssh -C -o ServerAliveInterval=255 -t username@address.com')

def test_ssh_command_no_user():
    config = Config({
        'ssh': {
            'options': '-C -o ServerAliveInterval=255'
        }
    })
    sshObj = Ssh(config, instance)
    assert_command_results(sshObj.command, 'ssh -C -o ServerAliveInterval=255 -t address.com')

def test_ssh_command_bastion_options():
    config = Config({
        'bastion': {
            'address': 'bastion.com',
            'options': '-C -o ServerAliveInterval=255'
        }
    })
    sshObj = Ssh(config, instance)
    assert_command_results(sshObj.command, "ssh -o ProxyCommand='ssh -C -o ServerAliveInterval=255 -W %h:%p bastion.com' -t address.com")

def test_ssh_command_bastion_user():
    config = Config({
        'bastion': {
            'address': 'bastion.com',
            'user': 'bastionuser'
        }
    })
    sshObj = Ssh(config, instance)
    assert_command_results(sshObj.command, "ssh -o ProxyCommand='ssh -W %h:%p bastionuser@bastion.com' -t address.com")

def test_ssh_command_bastion_missing_address():
    config = Config({
        'bastion': {}
    })
    with pytest.raises(MissingBastionHost):
        sshObj = Ssh(config, instance)
        sshObj.command

def test_ssh_command_null_instance():
    config = Config({})
    with pytest.raises(BadInstanceError):
        sshObj = Ssh(config, None)
        sshObj.command

def test_ssh_command_null_config():
    with pytest.raises(BadConfigError):
        sshObj = Ssh(None, instance)
        sshObj.command