Skip to content
Snippets Groups Projects
Commit 07633cce authored by JORGE SASIAIN's avatar JORGE SASIAIN
Browse files

NAT-152: Add test. Comment out unused code

parent daa2edc7
No related branches found
No related tags found
1 merge request!9Ipam service
This commit is part of merge request !9. Comments created here will be created in the context of that merge request.
Hide whitespace changes
Inline Side-by-side
gso/services/_ipam.py
+ 4
− 4
View file @ 07633cce
import ipaddress import ipaddress
import json
import requests import requests
from enum import Enum from enum import Enum
from pydantic import BaseSettings from pydantic import BaseSettings
...@@ -439,7 +438,7 @@ def allocate_service_host(hostname=None, ...@@ -439,7 +438,7 @@ def allocate_service_host(hostname=None,
Below methods are not used for supported outside calls Below methods are not used for supported outside calls
""" """
'''
def _find_containers(network=None, ip_version=4): def _find_containers(network=None, ip_version=4):
""" """
If network is not None, find that container. If network is not None, find that container.
...@@ -588,8 +587,8 @@ def _get_network_usage_status(network): ...@@ -588,8 +587,8 @@ def _get_network_usage_status(network):
assert r.status_code >= 200 and r.status_code < 300, \ assert r.status_code >= 200 and r.status_code < 300, \
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}" f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
return r.json() return r.json()
'''
'''
if __name__ == '__main__': if __name__ == '__main__':
while True: while True:
print("1. Find all containers") print("1. Find all containers")
...@@ -679,3 +678,4 @@ if __name__ == '__main__': ...@@ -679,3 +678,4 @@ if __name__ == '__main__':
else: else:
print("Invalid choice. Please try again.") print("Invalid choice. Please try again.")
'''
gso/services/ipam.py
+ 4
− 0
View file @ 07633cce
...@@ -55,6 +55,7 @@ def new_service_host(hostname, ...@@ -55,6 +55,7 @@ def new_service_host(hostname,
extattrs=extattrs) extattrs=extattrs)
'''
if __name__ == '__main__': if __name__ == '__main__':
# sample call flow to allocate two loopback interfaces and a trunk service # sample call flow to allocate two loopback interfaces and a trunk service
# new_service_host can be called passing networks or addresses # new_service_host can be called passing networks or addresses
...@@ -74,6 +75,8 @@ if __name__ == '__main__': ...@@ -74,6 +75,8 @@ if __name__ == '__main__':
) )
lo1_v4_host_address = lo1_service_networks.v4.network_address lo1_v4_host_address = lo1_service_networks.v4.network_address
lo1_v6_host_address = lo1_service_networks.v6.network_address lo1_v6_host_address = lo1_service_networks.v6.network_address
print(lo1_v4_host_address)
print(lo1_v6_host_address)
lo1_host_addresses = HostAddresses(v4=lo1_v4_host_address, lo1_host_addresses = HostAddresses(v4=lo1_v4_host_address,
v6=lo1_v6_host_address) v6=lo1_v6_host_address)
new_service_host(hostname=hostname_A, new_service_host(hostname=hostname_A,
...@@ -107,3 +110,4 @@ if __name__ == '__main__': ...@@ -107,3 +110,4 @@ if __name__ == '__main__':
new_service_host(hostname=hostname_B, new_service_host(hostname=hostname_B,
service_type='TRUNK', service_type='TRUNK',
service_networks=trunk12_service_networks) service_networks=trunk12_service_networks)
'''
setup.py
+ 2
− 1
View file @ 07633cce
...@@ -9,7 +9,8 @@ setup( ...@@ -9,7 +9,8 @@ setup(
url=('https://gitlab.geant.org/goat/geant-service-orchestrator'), url=('https://gitlab.geant.org/goat/geant-service-orchestrator'),
packages=find_packages(), packages=find_packages(),
install_requires=[ install_requires=[
'requests',
'orchestrator-core==1.0.0', 'orchestrator-core==1.0.0',
'pydantic',
'requests',
] ]
) )
test/test_ipam.py 0 → 100644
+ 194
− 0
View file @ 07633cce
import contextlib
import ipaddress
import json
import os
import socket
import pytest
import re
import responses
import tempfile
from gso.services import ipam
@pytest.fixture(scope='session')
def configuration_data():
with contextlib.closing(
socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
s.bind(('', 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
yield {
"GENERAL": {
"public_hostname": "https://gap.geant.org"
},
"RESOURCE_MANAGER_API_PREFIX": "http://localhost:44444",
"IPAM": {
"INFOBLOX": {
"scheme": "https",
"wapi_version": "v2.12",
"host": "10.0.0.1",
"username": "robot-user",
"password": "robot-user-password"
},
"LO": {
"V4": {"containers": ["10.255.255.0/24"], "mask": 32},
"V6": {"containers": ["dead:beef::/64"], "mask": 128},
"domain_name": ".lo"
},
"TRUNK": {
"V4": {
"containers": ["10.255.255.0/24", "10.255.254.0/24"],
"mask": 31
},
"V6": {
"containers": ["dead:beef::/64", "dead:beee::/64"],
"mask": 126
},
"domain_name": ".trunk"
},
"GEANT_IP": {
"V4": {
"containers": ["10.255.255.0/24", "10.255.254.0/24"],
"mask": 31
},
"V6": {
"containers": ["dead:beef::/64", "dead:beee::/64"],
"mask": 126
},
"domain_name": ".geantip"
}
},
"PROVISIONING_PROXY": {
"scheme": "https",
"api_base": "localhost:44444",
"auth": "Bearer <token>",
"api_version": 1123
}
}
@pytest.fixture(scope='session')
def data_config_filename(configuration_data):
file_name = os.path.join(
tempfile.gettempdir(), os.urandom(24).hex())
open(file_name, 'x').close()
with open(file_name, 'wb') as f:
f.write(json.dumps(configuration_data).encode('utf-8'))
f.flush()
os.environ['OSS_PARAMS_FILENAME'] = f.name
yield f.name
@responses.activate
def test_new_service_networks(data_config_filename, service_type='LO'):
responses.add(
method=responses.POST,
url=re.compile(r'.*/wapi.*/network.*'),
json={
'_ref': 'network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.20/32/default', # noqa: E501
'network': '10.255.255.20/32'
}
)
responses.add(
method=responses.POST,
url=re.compile(r'.*/wapi.*/ipv6network.*'),
json={
'_ref': 'ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default', # noqa: E501
'network': 'dead:beef::18/128'
}
)
service_networks = ipam.new_service_networks(service_type='LO')
assert service_networks == ipam.ServiceNetworks(
v4=ipaddress.ip_network('10.255.255.20/32'),
v6=ipaddress.ip_network('dead:beef::18/128')
)
@responses.activate
def test_new_service_host(data_config_filename, service_type='LO'):
responses.add(
method=responses.POST,
url=re.compile(r'.*/wapi.*/record:host$'),
json='record:host/ZG5zLmhvc3QkLm5vbl9ETlNfaG9zdF9yb290LjAuMTY4MzcwNTU4MzY3MC5nc28udGVzdA:test.lo/%20' # noqa: E501
)
responses.add(
method=responses.POST,
url=re.compile(r'.*/wapi.*/record:a$'),
json='record:a/ZG5zLmJpbmRfYSQuX2RlZmF1bHQuZ3NvLHRlc3QsMTAuMjU1LjI1NS44:test.lo/default' # noqa: E501
)
responses.add(
method=responses.POST,
url=re.compile(r'.*/wapi.*/record:aaaa$'),
json='record:aaaa/ZG5zLmJpbmRfYSQuX2RlZmF1bHQuZ3NvLHRlc3QsMTAuMjU1LjI1NS44:test.lo/default' # noqa: E501
)
responses.add(
method=responses.GET,
url=re.compile(r'.*/wapi.*/network.*'),
json=[
{
"_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.20/32/default", # noqa: E501
"network": "10.255.255.20/32",
"network_view": "default"
}
]
)
responses.add(
method=responses.GET,
url=re.compile(r'.*/wapi.*/ipv6network.*'),
json=[
{
"_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default", # noqa: E501
"network": "dead:beef::18/128",
"network_view": "default"
}
]
)
responses.add(
method=responses.POST,
url=re.compile(r'.*/wapi.*/network.*/.*?_function=next_available_ip&num=1.*'), # noqa: E501
json={'ips': ['10.255.255.20']}
)
responses.add(
method=responses.POST,
url=re.compile(r'.*/wapi.*/ipv6network.*/.*?_function=next_available_ip&num=1.*'), # noqa: E501
json={'ips': ['dead:beef::18']}
)
service_hosts = ipam.new_service_host(
hostname='test',
service_type='LO',
host_addresses=ipam.HostAddresses(
v4=ipaddress.ip_address('10.255.255.20'),
v6=ipaddress.ip_address('dead:beef::18')
)
)
assert service_hosts == ipam.HostAddresses(
v4=ipaddress.ip_address('10.255.255.20'),
v6=ipaddress.ip_address('dead:beef::18')
)
service_hosts = ipam.new_service_host(
hostname='test',
service_type='LO',
service_networks=ipam.ServiceNetworks(
v4=ipaddress.ip_network('10.255.255.20/32'),
v6=ipaddress.ip_network('dead:beef::18/128')
)
)
assert service_hosts == ipam.HostAddresses(
v4=ipaddress.ip_address('10.255.255.20'),
v6=ipaddress.ip_address('dead:beef::18')
)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment