nomilo/e2e/zones.py

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')