test_octodns_provider_digitalocean.py 文件源码

def test_populate(self):
        provider = DigitalOceanProvider('test', 'token')

        # Bad auth
        with requests_mock() as mock:
            mock.get(ANY, status_code=401,
                     text='{"id":"unauthorized",'
                     '"message":"Unable to authenticate you."}')

            with self.assertRaises(Exception) as ctx:
                zone = Zone('unit.tests.', [])
                provider.populate(zone)
            self.assertEquals('Unauthorized', ctx.exception.message)

        # General error
        with requests_mock() as mock:
            mock.get(ANY, status_code=502, text='Things caught fire')

            with self.assertRaises(HTTPError) as ctx:
                zone = Zone('unit.tests.', [])
                provider.populate(zone)
            self.assertEquals(502, ctx.exception.response.status_code)

        # Non-existant zone doesn't populate anything
        with requests_mock() as mock:
            mock.get(ANY, status_code=404,
                     text='{"id":"not_found","message":"The resource you '
                     'were accessing could not be found."}')

            zone = Zone('unit.tests.', [])
            provider.populate(zone)
            self.assertEquals(set(), zone.records)

        # No diffs == no changes
        with requests_mock() as mock:
            base = 'https://api.digitalocean.com/v2/domains/unit.tests/' \
                'records?page='
            with open('tests/fixtures/digitalocean-page-1.json') as fh:
                mock.get('{}{}'.format(base, 1), text=fh.read())
            with open('tests/fixtures/digitalocean-page-2.json') as fh:
                mock.get('{}{}'.format(base, 2), text=fh.read())

            zone = Zone('unit.tests.', [])
            provider.populate(zone)
            self.assertEquals(12, len(zone.records))
            changes = self.expected.changes(zone, provider)
            self.assertEquals(0, len(changes))

        # 2nd populate makes no network calls/all from cache
        again = Zone('unit.tests.', [])
        provider.populate(again)
        self.assertEquals(12, len(again.records))

        # bust the cache
        del provider._zone_records[zone.name]