nomilo/e2e/zones.py

159 lines
5.2 KiB
Python
Raw Normal View History

2022-03-04 12:08:47 +00:00
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,
2022-03-04 20:55:27 +00:00
RecordList,
UpdateRecordsRequest,
2022-04-23 21:30:20 +00:00
CreateZoneRequest,
2022-03-04 12:08:47 +00:00
)
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')
2022-03-04 20:55:27 +00:00
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'
)
2022-03-05 12:07:51 +00:00
self.api.zones_zone_records_post(zone='example.com.', record_list=RecordList([old_record]))
2022-03-04 20:55:27 +00:00
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
2022-03-05 12:07:51 +00:00
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
2022-04-23 21:30:20 +00:00
self.assertFalse(found, msg='Delete record found in zone records')