diff --git a/gso/services/_ipam.py b/gso/services/_ipam.py
index 01dc0b0957b175ecb9cc1509db241e269c0cfa38..2707a53ddc0ef4c2111baca5b8ac640f80a9b614 100644
--- a/gso/services/_ipam.py
+++ b/gso/services/_ipam.py
@@ -1,5 +1,4 @@
import ipaddress
-import json
import requests
from enum import Enum
from pydantic import BaseSettings
@@ -439,7 +438,7 @@ def allocate_service_host(hostname=None,
Below methods are not used for supported outside calls
"""
-
+'''
def _find_containers(network=None, ip_version=4):
"""
If network is not None, find that container.
@@ -588,8 +587,8 @@ def _get_network_usage_status(network):
assert r.status_code >= 200 and r.status_code < 300, \
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
return r.json()
-
-
+'''
+'''
if __name__ == '__main__':
while True:
print("1. Find all containers")
@@ -679,3 +678,4 @@ if __name__ == '__main__':
else:
print("Invalid choice. Please try again.")
+'''
diff --git a/gso/services/ipam.py b/gso/services/ipam.py
index 44b75bacbe6ff9387f742e9ecbde0e9e5b4a2029..c41a1239a14506e3104aaad5d5e4dcdfadd4dcae 100644
--- a/gso/services/ipam.py
+++ b/gso/services/ipam.py
@@ -55,6 +55,7 @@ def new_service_host(hostname,
extattrs=extattrs)
+'''
if __name__ == '__main__':
# sample call flow to allocate two loopback interfaces and a trunk service
# new_service_host can be called passing networks or addresses
@@ -74,6 +75,8 @@ if __name__ == '__main__':
)
lo1_v4_host_address = lo1_service_networks.v4.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,
v6=lo1_v6_host_address)
new_service_host(hostname=hostname_A,
@@ -107,3 +110,4 @@ if __name__ == '__main__':
new_service_host(hostname=hostname_B,
service_type='TRUNK',
service_networks=trunk12_service_networks)
+'''
diff --git a/requirements.txt b/requirements.txt
index 243292eec5c1bf77b11fc4166714d056f98fd313..3ac3e8394ad7a4649d8b87cd724f0d44ed8c1e03 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,6 @@
orchestrator-core==1.0.0
+pydantic
requests
pytest
+responses
diff --git a/setup.py b/setup.py
index 50d3dc63c04f6f68ec902ab6478942f252c4dc40..7062c773b0d214c582e7c5b9db49da806d1eae59 100644
--- a/setup.py
+++ b/setup.py
@@ -9,7 +9,8 @@ setup(
url=('https://gitlab.geant.org/goat/geant-service-orchestrator'),
packages=find_packages(),
install_requires=[
- 'requests',
'orchestrator-core==1.0.0',
+ 'pydantic',
+ 'requests',
]
)
diff --git a/test/test_ipam.py b/test/test_ipam.py
new file mode 100644
index 0000000000000000000000000000000000000000..6c7c5dd14bc86b995a120b860af0b8c611a1e4cc
--- /dev/null
+++ b/test/test_ipam.py
@@ -0,0 +1,194 @@
+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')
+ )