How to use the octodns.provider.yaml.YamlProvider function in octodns

def make_expected(self):
        expected = Zone('unit.tests.', [])
        source = YamlProvider('test', join(dirname(__file__), 'config'))
        source.populate(expected)
        return expected

self.assertEquals(21, tc)

            # Include meta
            tc = Manager(get_config_filename('simple.yaml'), max_workers=1,
                         include_meta=True) \
                .sync(dry_run=False, force=True)
            self.assertEquals(25, tc)

            # Cautious
            tc = Manager(get_config_filename('simple.yaml'), max_workers=1,
                         include_meta=True) \
                .sync(dry_run=False, cautious=True)
            # Same changes
            self.assertEquals(25, tc)
            # But we expect all of their TTLs to be 60s
            source = YamlProvider('verify', tmpdir.dirname)
            found = Zone('unit.tests.', [])
            source.populate(found)
            for record in found.records:
                self.assertEquals(60, record.ttl)

def test_unsorted(self):
        source = YamlProvider('test', join(dirname(__file__), 'config'))

        zone = Zone('unordered.', [])

        with self.assertRaises(ConstructorError):
            source.populate(zone)

        source = YamlProvider('test', join(dirname(__file__), 'config'),
                              enforce_order=False)
        # no exception
        source.populate(zone)
        self.assertEqual(2, len(zone.records))

def test_subzone_handling(self):
        source = YamlProvider('test', join(dirname(__file__), 'config'))

        # If we add `sub` as a sub-zone we'll reject `www.sub`
        zone = Zone('unit.tests.', ['sub'])
        with self.assertRaises(SubzoneRecordException) as ctx:
            source.populate(zone)
        self.assertEquals('Record www.sub.unit.tests. is under a managed '
                          'subzone', ctx.exception.message)

from mock import Mock, call
from os.path import dirname, join
from requests import HTTPError
from requests_mock import ANY, mock as requests_mock
from unittest import TestCase

from octodns.record import Record
from octodns.provider.dnsimple import DnsimpleClientNotFound, DnsimpleProvider
from octodns.provider.yaml import YamlProvider
from octodns.zone import Zone


class TestDnsimpleProvider(TestCase):
    expected = Zone('unit.tests.', [])
    source = YamlProvider('test', join(dirname(__file__), 'config'))
    source.populate(expected)

    # Our test suite differs a bit, add our NS and remove the simple one
    expected.add_record(Record.new(expected, 'under', {
        'ttl': 3600,
        'type': 'NS',
        'values': [
            'ns1.unit.tests.',
            'ns2.unit.tests.',
        ]
    }))
    for record in list(expected.records):
        if record.name == 'sub' and record._type == 'NS':
            expected._remove_record(record)
            break

def test_empty(self):
        source = YamlProvider('test', join(dirname(__file__), 'config'))

        zone = Zone('empty.', [])

        # without it we see everything
        source.populate(zone)
        self.assertEquals(0, len(zone.records))

def test_unsorted(self):
        source = YamlProvider('test', join(dirname(__file__), 'config'))

        zone = Zone('unordered.', [])

        with self.assertRaises(ConstructorError):
            source.populate(zone)

        source = YamlProvider('test', join(dirname(__file__), 'config'),
                              enforce_order=False)
        # no exception
        source.populate(zone)
        self.assertEqual(2, len(zone.records))

from os.path import dirname, join

from requests_mock import ANY, mock as requests_mock
from unittest import TestCase

from octodns.provider.mythicbeasts import MythicBeastsProvider, \
    add_trailing_dot, remove_trailing_dot
from octodns.provider.yaml import YamlProvider
from octodns.zone import Zone
from octodns.record import Create, Update, Delete, Record


class TestMythicBeastsProvider(TestCase):
    expected = Zone('unit.tests.', [])
    source = YamlProvider('test_expected', join(dirname(__file__), 'config'))
    source.populate(expected)

    # Dump anything we don't support from expected
    for record in list(expected.records):
        if record._type not in MythicBeastsProvider.SUPPORTS:
            expected._remove_record(record)

    def test_trailing_dot(self):
        with self.assertRaises(AssertionError) as err:
            add_trailing_dot('unit.tests.')
        self.assertEquals('Value already has trailing dot',
                          err.exception.message)

        with self.assertRaises(AssertionError) as err:
            remove_trailing_dot('unit.tests')
        self.assertEquals('Value already missing trailing dot',

from mock import Mock, call
from os.path import dirname, join
from requests import HTTPError
from requests_mock import ANY, mock as requests_mock
from unittest import TestCase

from octodns.record import Record, Update
from octodns.provider.base import Plan
from octodns.provider.cloudflare import CloudflareProvider
from octodns.provider.yaml import YamlProvider
from octodns.zone import Zone


class TestCloudflareProvider(TestCase):
    expected = Zone('unit.tests.', [])
    source = YamlProvider('test', join(dirname(__file__), 'config'))
    source.populate(expected)

    # Our test suite differs a bit, add our NS and remove the simple one
    expected.add_record(Record.new(expected, 'under', {
        'ttl': 3600,
        'type': 'NS',
        'values': [
            'ns1.unit.tests.',
            'ns2.unit.tests.',
        ]
    }))
    for record in list(expected.records):
        if record.name == 'sub' and record._type == 'NS':
            expected._remove_record(record)
            break