mirror of
https://github.com/dcarrillo/digaws.git
synced 2025-07-02 06:19:26 +00:00
First commit
This commit is contained in:
134
tests/__init__.py
Normal file
134
tests/__init__.py
Normal file
@ -0,0 +1,134 @@
|
||||
import ipaddress
|
||||
|
||||
AWS_IP_RANGES = '''
|
||||
{
|
||||
"syncToken": "1608245058",
|
||||
"createDate": "2020-12-17-22-44-18",
|
||||
"prefixes": [
|
||||
{
|
||||
"ip_prefix": "52.93.178.234/32",
|
||||
"region": "us-west-1",
|
||||
"service": "AMAZON",
|
||||
"network_border_group": "us-west-1"
|
||||
},
|
||||
{
|
||||
"ip_prefix": "52.94.76.0/22",
|
||||
"region": "us-west-2",
|
||||
"service": "AMAZON",
|
||||
"network_border_group": "us-west-2"
|
||||
}
|
||||
],
|
||||
"ipv6_prefixes": [
|
||||
{
|
||||
"ipv6_prefix": "2600:1f00:c000::/40",
|
||||
"region": "us-west-1",
|
||||
"service": "AMAZON",
|
||||
"network_border_group": "us-west-1"
|
||||
},
|
||||
{
|
||||
"ipv6_prefix": "2600:1f01:4874::/47",
|
||||
"region": "us-west-2",
|
||||
"service": "AMAZON",
|
||||
"network_border_group": "us-west-2"
|
||||
},
|
||||
{
|
||||
"ipv6_prefix": "2600:1f14:fff:f800::/53",
|
||||
"region": "us-west-2",
|
||||
"service": "ROUTE53_HEALTHCHECKS",
|
||||
"network_border_group": "us-west-2"
|
||||
},
|
||||
{
|
||||
"ipv6_prefix": "2600:1f14::/35",
|
||||
"region": "us-west-2",
|
||||
"service": "EC2",
|
||||
"network_border_group": "us-west-2"
|
||||
}
|
||||
]
|
||||
}
|
||||
'''
|
||||
AWS_IPV4_RANGES_OBJ = [
|
||||
{
|
||||
'ip_prefix': ipaddress.IPv4Network('52.93.178.234/32'),
|
||||
'region': 'us-west-1',
|
||||
'service': 'AMAZON',
|
||||
'network_border_group': 'us-west-1'
|
||||
},
|
||||
{
|
||||
'ip_prefix': ipaddress.IPv4Network('52.94.76.0/22'),
|
||||
'region': 'us-west-2',
|
||||
'service': 'AMAZON',
|
||||
'network_border_group': 'us-west-2'
|
||||
}
|
||||
]
|
||||
AWS_IPV6_RANGES_OBJ = [
|
||||
{
|
||||
'ipv6_prefix': ipaddress.IPv6Network('2600:1f00:c000::/40'),
|
||||
'region': 'us-west-1',
|
||||
'service': 'AMAZON',
|
||||
'network_border_group': 'us-west-1'
|
||||
},
|
||||
{
|
||||
'ipv6_prefix': ipaddress.IPv6Network('2600:1f01:4874::/47'),
|
||||
'region': 'us-west-2',
|
||||
'service': 'AMAZON',
|
||||
'network_border_group': 'us-west-2'
|
||||
},
|
||||
{
|
||||
'ipv6_prefix': ipaddress.IPv6Network('2600:1f14:fff:f800::/53'),
|
||||
'region': 'us-west-2',
|
||||
'service': 'ROUTE53_HEALTHCHECKS',
|
||||
'network_border_group': 'us-west-2'
|
||||
},
|
||||
{
|
||||
'ipv6_prefix': ipaddress.IPv6Network('2600:1f14::/35'),
|
||||
'region': 'us-west-2',
|
||||
'service': 'EC2',
|
||||
'network_border_group': 'us-west-2'
|
||||
}
|
||||
]
|
||||
LAST_MODIFIED_TIME = 'Thu, 17 Dec 2020 23:22:33 GMT'
|
||||
|
||||
RESPONSE_PLAIN_PRINT = '''Prefix: 52.94.76.0/22
|
||||
Region: us-west-2
|
||||
Service: AMAZON
|
||||
Network border group: us-west-2
|
||||
|
||||
'''
|
||||
|
||||
RESPONSE_JSON_PRINT = '''[
|
||||
{
|
||||
"ipv6_prefix": "2600:1f14:fff:f800::/53",
|
||||
"region": "us-west-2",
|
||||
"service": "ROUTE53_HEALTHCHECKS",
|
||||
"network_border_group": "us-west-2"
|
||||
},
|
||||
{
|
||||
"ipv6_prefix": "2600:1f14::/35",
|
||||
"region": "us-west-2",
|
||||
"service": "EC2",
|
||||
"network_border_group": "us-west-2"
|
||||
}
|
||||
]
|
||||
'''
|
||||
|
||||
RESPONSE_JSON_JOINED_PRINT = '''[
|
||||
{
|
||||
"ip_prefix": "52.94.76.0/22",
|
||||
"region": "us-west-2",
|
||||
"service": "AMAZON",
|
||||
"network_border_group": "us-west-2"
|
||||
},
|
||||
{
|
||||
"ipv6_prefix": "2600:1f14:fff:f800::/53",
|
||||
"region": "us-west-2",
|
||||
"service": "ROUTE53_HEALTHCHECKS",
|
||||
"network_border_group": "us-west-2"
|
||||
},
|
||||
{
|
||||
"ipv6_prefix": "2600:1f14::/35",
|
||||
"region": "us-west-2",
|
||||
"service": "EC2",
|
||||
"network_border_group": "us-west-2"
|
||||
}
|
||||
]
|
||||
'''
|
77
tests/test_dig_aws.py
Normal file
77
tests/test_dig_aws.py
Normal file
@ -0,0 +1,77 @@
|
||||
import json
|
||||
import sys
|
||||
|
||||
import digaws.digaws as digaws
|
||||
from digaws import __description__, __version__
|
||||
|
||||
import pytest
|
||||
|
||||
import tests
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_dig():
|
||||
return digaws.DigAWS(json.loads(tests.AWS_IP_RANGES))
|
||||
|
||||
|
||||
def test_cli(capsys):
|
||||
sys.argv = ['digaws', '-h']
|
||||
try:
|
||||
digaws.main()
|
||||
except SystemExit as e:
|
||||
out, _ = capsys.readouterr()
|
||||
assert __description__ in out
|
||||
assert e.code == 0
|
||||
|
||||
|
||||
def test_cli_version(capsys, mocker):
|
||||
sys.argv = ['digaws', '--version']
|
||||
try:
|
||||
digaws.main()
|
||||
except SystemExit as e:
|
||||
out, _ = capsys.readouterr()
|
||||
assert out == f'digaws {__version__}\n'
|
||||
assert e.code == 0
|
||||
|
||||
|
||||
def test_cli_invocation(capsys, mocker):
|
||||
sys.argv = ['digaws', '52.94.76.0/22', '2600:1f14:fff:f810:a1c1:f507:a2d1:2dd8',
|
||||
'--output', 'json']
|
||||
mocker.patch('digaws.digaws.get_aws_ip_ranges', return_value=json.loads(tests.AWS_IP_RANGES))
|
||||
digaws.main()
|
||||
out, _ = capsys.readouterr()
|
||||
|
||||
assert out == tests.RESPONSE_JSON_JOINED_PRINT
|
||||
|
||||
|
||||
def test_dig_aws_construct(test_dig):
|
||||
assert test_dig.ip_prefixes == tests.AWS_IPV4_RANGES_OBJ
|
||||
assert test_dig.ipv6_prefixes == tests.AWS_IPV6_RANGES_OBJ
|
||||
|
||||
|
||||
def test_lookup(test_dig):
|
||||
assert str(test_dig._lookup_data('52.94.76.1')[0]['ip_prefix']) == '52.94.76.0/22'
|
||||
assert str(test_dig._lookup_data('52.94.76.0/24')[0]['ip_prefix']) == '52.94.76.0/22'
|
||||
|
||||
input = '2600:1f14:fff:f810:a1c1:f507:a2d1:2dd8'
|
||||
assert str(test_dig._lookup_data(input)[0]['ipv6_prefix']) == '2600:1f14:fff:f800::/53'
|
||||
assert str(test_dig._lookup_data(input)[1]['ipv6_prefix']) == '2600:1f14::/35'
|
||||
assert str(test_dig._lookup_data('2600:1f14::/36')[0]['ipv6_prefix']) == '2600:1f14::/35'
|
||||
|
||||
with pytest.raises(ValueError) as e:
|
||||
test_dig.lookup('what are you talking about')
|
||||
assert e.startswith('Wrong IP or CIDR format')
|
||||
|
||||
|
||||
def test_response_plain_print(test_dig, capsys):
|
||||
test_dig.lookup('52.94.76.0/22').plain_print()
|
||||
out, _ = capsys.readouterr()
|
||||
|
||||
assert out == tests.RESPONSE_PLAIN_PRINT
|
||||
|
||||
|
||||
def test_response_json_print(test_dig, capsys):
|
||||
test_dig.lookup('2600:1f14:fff:f810:a1c1:f507:a2d1:2dd8').json_print()
|
||||
out, _ = capsys.readouterr()
|
||||
|
||||
assert out == tests.RESPONSE_JSON_PRINT
|
87
tests/test_get_aws_ip_ranges_json.py
Normal file
87
tests/test_get_aws_ip_ranges_json.py
Normal file
@ -0,0 +1,87 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
import digaws.digaws as digaws
|
||||
|
||||
import pytest
|
||||
|
||||
import requests
|
||||
|
||||
import tests
|
||||
|
||||
|
||||
class MockGetResponse:
|
||||
text = tests.AWS_IP_RANGES
|
||||
status_code = 200
|
||||
|
||||
@staticmethod
|
||||
def json():
|
||||
return json.loads(tests.AWS_IP_RANGES)
|
||||
|
||||
|
||||
def mock_get(*args, **kwargs):
|
||||
return MockGetResponse()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def create_cache_dir(fs):
|
||||
digaws.CACHE_DIR.mkdir(parents=True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('fs', [[None, [digaws]]], indirect=True)
|
||||
def test_get_aws_ip_ranges_cached_valid_file(mocker, fs, create_cache_dir) -> None:
|
||||
with open(digaws.CACHE_FILE, 'w') as out:
|
||||
out.write(tests.AWS_IP_RANGES)
|
||||
|
||||
response = requests.Response
|
||||
response.status_code = 304
|
||||
mocker.patch('requests.get', return_value=response)
|
||||
|
||||
result = digaws.get_aws_ip_ranges()
|
||||
|
||||
assert result['syncToken'] == '1608245058'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('fs', [[None, [digaws]]], indirect=True)
|
||||
def test_get_aws_ip_ranges_cached_invalid_file(mocker, fs, create_cache_dir) -> None:
|
||||
with open(digaws.CACHE_FILE, 'w'):
|
||||
pass
|
||||
|
||||
response = requests.Response
|
||||
response.status_code = 304
|
||||
mocker.patch('requests.get', return_value=response)
|
||||
|
||||
with pytest.raises(digaws.CachedFileException):
|
||||
digaws.get_aws_ip_ranges()
|
||||
|
||||
|
||||
@pytest.mark.parametrize('fs', [[None, [digaws]]], indirect=True)
|
||||
def test_get_aws_ip_ranges_cached_deprecated_file(monkeypatch, fs, create_cache_dir) -> None:
|
||||
with open(digaws.CACHE_FILE, 'w'):
|
||||
pass
|
||||
digaws.CACHE_FILE.touch()
|
||||
os.utime(digaws.CACHE_FILE, times=(0, 0))
|
||||
|
||||
monkeypatch.setattr(requests, 'get', mock_get)
|
||||
result = digaws.get_aws_ip_ranges()
|
||||
|
||||
assert result['syncToken'] == '1608245058'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('fs', [[None, [digaws]]], indirect=True)
|
||||
def test_get_aws_ip_ranges_no_file(monkeypatch, fs, create_cache_dir) -> None:
|
||||
monkeypatch.setattr(requests, 'get', mock_get)
|
||||
result = digaws.get_aws_ip_ranges()
|
||||
|
||||
assert result['syncToken'] == '1608245058'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('fs', [[None, [digaws]]], indirect=True)
|
||||
def test_get_aws_ip_ranges_invalid_status(mocker, fs, create_cache_dir) -> None:
|
||||
response = requests.Response
|
||||
response.status_code = 301
|
||||
mocker.patch('requests.get', return_value=response)
|
||||
|
||||
with pytest.raises(digaws.UnexpectedRequestException) as e:
|
||||
digaws.get_aws_ip_ranges()
|
||||
assert e.startswith('Unexpected response from')
|
Reference in New Issue
Block a user