159 lines
5.2 KiB
Python
159 lines
5.2 KiB
Python
from nomilo_client import ApiClient, Configuration
|
|
from nomilo_client.api.default_api import DefaultApi
|
|
from nomilo_client.models import (
|
|
TokenRequest,
|
|
RecordTypeSOA,
|
|
RecordTypeAAAA,
|
|
RecordTypeCNAME,
|
|
RecordTypeNS,
|
|
RecordTypeTXT,
|
|
RecordList,
|
|
UpdateRecordsRequest,
|
|
CreateZoneRequest,
|
|
)
|
|
|
|
import logging
|
|
import string
|
|
import random
|
|
|
|
import unittest
|
|
import warnings
|
|
|
|
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
HOST = 'http://localhost:8000/api/v1'
|
|
USER='toto'
|
|
PASSWORD='supersecure'
|
|
|
|
|
|
def build_api(host: str):
|
|
conf = Configuration(host=HOST)
|
|
api_client = ApiClient(configuration=conf)
|
|
return DefaultApi(api_client)
|
|
|
|
def build_authenticated_api(host: str, token: TokenRequest):
|
|
auth_conf = Configuration(host=host, access_token=token.token)
|
|
api_client = ApiClient(configuration=auth_conf)
|
|
return DefaultApi(api_client)
|
|
|
|
def random_string(length):
|
|
return ''.join(random.choice(string.ascii_lowercase) for x in range(length))
|
|
|
|
def random_name(zone):
|
|
return '%s.%s' % (random_string(16), zone)
|
|
|
|
|
|
class TestZones(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
# Ignore warning about unclosed socket
|
|
warnings.filterwarnings(action="ignore", message="unclosed", category=ResourceWarning)
|
|
|
|
api = build_api(HOST)
|
|
token = api.users_me_token_post(token_request=TokenRequest(username=USER,password=PASSWORD))
|
|
cls.api = build_authenticated_api(HOST, token)
|
|
|
|
def test_get_zones(self):
|
|
zones = self.api.zones_get()
|
|
zone_name = zones.value[0].name
|
|
self.assertEqual(zone_name, 'example.com.')
|
|
|
|
def test_get_records(self):
|
|
records = self.api.zones_zone_records_get(zone='example.com.')
|
|
for record in records.value:
|
|
if type(record) is RecordTypeSOA:
|
|
with self.subTest(type='soa'):
|
|
self.assertEqual(record.name, 'example.com.')
|
|
|
|
if type(record) is RecordTypeAAAA:
|
|
with self.subTest(type='ns'):
|
|
self.assertEqual(record.name, 'srv1.example.com.')
|
|
self.assertEqual(record.address, '2001:db8:cafe:bc68::2')
|
|
|
|
if type(record) is RecordTypeCNAME:
|
|
with self.subTest(type='cname'):
|
|
self.assertEqual(record.name, 'www.example.com.')
|
|
self.assertEqual(record.target, 'srv1.example.com.')
|
|
|
|
if type(record) is RecordTypeNS:
|
|
with self.subTest(type='ns'):
|
|
self.assertEqual(record.name, 'example.com.')
|
|
self.assertEqual(record.target, 'ns.example.com.')
|
|
|
|
def test_create_records(self):
|
|
new_record = RecordTypeTXT(
|
|
_class='IN',
|
|
ttl=300,
|
|
name=random_name('example.com.'),
|
|
text=random_string(32),
|
|
type='TXT'
|
|
)
|
|
|
|
self.api.zones_zone_records_post(zone='example.com.', record_list=RecordList(value=[new_record]))
|
|
records = self.api.zones_zone_records_get(zone='example.com.')
|
|
found = False
|
|
for record in records.value:
|
|
if type(record) is RecordTypeTXT and record.name == new_record.name:
|
|
self.assertEqual(record.text, new_record.text, msg='New record does not have the expected value')
|
|
found = True
|
|
|
|
self.assertTrue(found, msg='New record not found in zone records')
|
|
|
|
def test_update_records(self):
|
|
name = random_name('example.com.')
|
|
old_record = RecordTypeTXT(
|
|
_class='IN',
|
|
ttl=300,
|
|
name=name,
|
|
text='old value',
|
|
type='TXT'
|
|
)
|
|
|
|
new_record = RecordTypeTXT(
|
|
_class='IN',
|
|
ttl=300,
|
|
name=name,
|
|
text='new value',
|
|
type='TXT'
|
|
)
|
|
|
|
self.api.zones_zone_records_post(zone='example.com.', record_list=RecordList([old_record]))
|
|
|
|
update_records_request = UpdateRecordsRequest(
|
|
old_records=RecordList([old_record]),
|
|
new_records=RecordList([new_record]),
|
|
)
|
|
|
|
self.api.zones_zone_records_put(zone='example.com.', update_records_request=update_records_request)
|
|
|
|
records = self.api.zones_zone_records_get(zone='example.com.')
|
|
found = False
|
|
for record in records.value:
|
|
if type(record) is RecordTypeTXT and record.name == name:
|
|
self.assertEqual(record.text, new_record.text, msg='New record does not have the expected value')
|
|
found = True
|
|
|
|
self.assertTrue(found, msg='Updated record not found in zone records')
|
|
|
|
def test_delete_records(self):
|
|
name = random_name('example.com.')
|
|
record = RecordTypeTXT(
|
|
_class='IN',
|
|
ttl=300,
|
|
name=name,
|
|
text=random_string(32),
|
|
type='TXT'
|
|
)
|
|
|
|
self.api.zones_zone_records_post(zone='example.com.', record_list=RecordList([record]))
|
|
self.api.zones_zone_records_delete(zone='example.com.', record_list=RecordList([record]))
|
|
|
|
records = self.api.zones_zone_records_get(zone='example.com.')
|
|
found = False
|
|
for record in records.value:
|
|
if type(record) is RecordTypeTXT and record.name == name:
|
|
found = True
|
|
|
|
self.assertFalse(found, msg='Delete record found in zone records')
|