How to use the octodns.record.Delete function in octodns

To help you get started, we’ve selected a few octodns examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github github / octodns / tests / test_octodns_provider_route53.py View on Github external
'HealthCheckConfig': {
                    'Type': 'HTTPS',
                    'FullyQualifiedDomainName': 'unit.tests',
                    'IPAddress': '2.2.3.4',
                },
                'HealthCheckVersion': 2,
            }],
            'IsTruncated': False,
            'MaxItems': '100',
            'Marker': '',
        })
        extra = provider._extra_changes(existing, [])
        self.assertEquals(1, len(extra))
        stubber.assert_no_pending_responses()

        for change in (Create(record), Update(record, record), Delete(record)):
            extra = provider._extra_changes(existing, [change])
            self.assertEquals(0, len(extra))
            stubber.assert_no_pending_responses()
github github / octodns / tests / test_octodns_plan.py View on Github external
'geo': {
        'AF': ['5.5.5.5'],
        'NA-US': ['6.6.6.6']
    },
    'ttl': 300,
    'type': 'A',
    # This leaves one, swaps ones, and adds one
    'values': ['2.2.2.2', '3.3.3.3', '4.4.4.4'],
}, simple)
create = Create(Record.new(zone, 'b', {
    'ttl': 60,
    'type': 'CNAME',
    'value': 'foo.unit.tests.'
}, simple))
update = Update(existing, new)
delete = Delete(new)
changes = [create, delete, update]
plans = [
    (simple, Plan(zone, zone, changes)),
    (simple, Plan(zone, zone, changes)),
]


class TestPlanHtml(TestCase):
    log = getLogger('TestPlanHtml')

    def test_empty(self):
        out = StringIO()
        PlanHtml('html').run([], fh=out)
        self.assertEquals('<b>No changes were planned</b>', out.getvalue())

    def test_simple(self):
github github / octodns / tests / test_octodns_plan.py View on Github external
'geo': {
                'AF': ['5.5.5.5'],
                'NA-US': ['6.6.6.6']
            },
            'ttl': 300,
            'type': 'A',
            # This leaves one, swaps ones, and adds one
            'values': ['2.2.2.2', '3.3.3.3', '4.4.4.4'],
        }, simple)
        create = Create(Record.new(zone, 'b', {
            'ttl': 3600,
            'type': 'CNAME',
            'value': 'foo.unit.tests.'
        }, simple))
        update = Update(existing, new)
        delete = Delete(new)
        changes = [create, delete, update]
        plan = Plan(zone, zone, changes)

        self.assertEquals(3600, create.new.ttl)
        self.assertEquals(300, update.new.ttl)
        plan.make_cautious()
        self.assertEquals(60, create.new.ttl)
        self.assertEquals(60, update.new.ttl)
github github / octodns / tests / test_octodns_provider_dyn.py View on Github external
def test_mod_geo_delete(self, mock):
        provider = DynProvider('test', 'cust', 'user', 'pass',
                               traffic_directors_enabled=True)

        td_mock = MagicMock()
        provider._traffic_directors = {
            'unit.tests.': {
                'A': td_mock,
            }
        }
        provider._mod_geo_Delete(None, Delete(self.geo_record))
        # delete called
        td_mock.delete.assert_called_once()
        # removed from cache
        self.assertFalse('A' in provider.traffic_directors['unit.tests.'])
github github / octodns / tests / test_octodns_provider_dyn.py View on Github external
mock.side_effect = [
            # get zone
            {'data': {}},
            # accept publish
            {'data': {}},
        ]
        desired = Zone('unit.tests.', [])
        geo = self.geo_record
        regular = self.regular_record

        changes = [
            Create(geo),
            Create(regular),
            Update(geo, geo),
            Update(regular, regular),
            Delete(geo),
            Delete(regular),
        ]
        plan = Plan(None, desired, changes, True)
        provider._apply(plan)
        mock.assert_has_calls([
            call('/Zone/unit.tests/', 'GET', {}),
            call('/Zone/unit.tests/', 'PUT', {'publish': True})
        ])
        # should have seen 1 call to each
        provider._mod_geo_Create.assert_called_once()
        provider._mod_geo_Update.assert_called_once()
        provider._mod_geo_Delete.assert_called_once()
        provider._mod_Create.assert_called_once()
        provider._mod_Update.assert_called_once()
        provider._mod_Delete.assert_called_once()
github github / octodns / tests / test_octodns_provider_ns1.py View on Github external
{'answer': ['3.4.5.6'], 'meta': {'country': ['US']}},
                    {'answer': ['4.5.6.7'],
                     'meta': {'iso_region_code': ['NA-US-WA']}},
                ],
                'ttl': 34,
            },
        ]
        nsone_zone.search = zone_search
        load_mock.side_effect = [nsone_zone, nsone_zone]
        plan = provider.plan(desired)
        self.assertEquals(3, len(plan.changes))
        # Shouldn't rely on order so just count classes
        classes = defaultdict(lambda: 0)
        for change in plan.changes:
            classes[change.__class__] += 1
        self.assertEquals(1, classes[Delete])
        self.assertEquals(2, classes[Update])
        # ugh, we need a mock record that can be returned from loadRecord for
        # the update and delete targets, we can add our side effects to that to
        # trigger rate limit handling
        mock_record = Mock()
        mock_record.update.side_effect = [
            RateLimitException('one', period=0),
            None,
            None,
        ]
        mock_record.delete.side_effect = [
            RateLimitException('two', period=0),
            None,
            None,
        ]
        nsone_zone.loadRecord.side_effect = [mock_record, mock_record,
github github / octodns / tests / test_octodns_provider_base.py View on Github external
# than MIN_EXISTING_RECORDS exist
        zone = Zone('unit.tests.', [])
        record = Record.new(zone, 'a', {
            'ttl': 30,
            'type': 'A',
            'value': '1.2.3.4',
        })

        for i in range(int(Plan.MIN_EXISTING_RECORDS)):
            zone.add_record(Record.new(zone, str(i), {
                            'ttl': 60,
                            'type': 'A',
                            'value': '2.3.4.5'
                            }))

        changes = [Delete(record)
                   for i in range(int(Plan.MIN_EXISTING_RECORDS *
                                      Plan.MAX_SAFE_DELETE_PCENT) + 1)]

        with self.assertRaises(UnsafePlan) as ctx:
            Plan(zone, zone, changes).raise_if_unsafe()

        self.assertTrue('Too many deletes' in ctx.exception.message)
github github / octodns / tests / test_octodns_zone.py View on Github external
target = SimpleProvider()

        # before == after -> no changes
        self.assertFalse(before.changes(after, target))

        # add a record, delete a record -> [Delete, Create]
        c = ARecord(before, 'c', {'ttl': 42, 'value': '1.1.1.1'})
        after.add_record(c)
        after._remove_record(b)
        self.assertEquals(after.records, set([a, c]))
        changes = before.changes(after, target)
        self.assertEquals(2, len(changes))
        for change in changes:
            if isinstance(change, Create):
                create = change
            elif isinstance(change, Delete):
                delete = change
        self.assertEquals(b, delete.existing)
        self.assertFalse(delete.new)
        self.assertEquals(c, create.new)
        self.assertFalse(create.existing)
        delete.__repr__()
        create.__repr__()

        after = Zone('unit.tests.', [])
        changed = ARecord(before, 'a', {'ttl': 42, 'value': '2.2.2.2'})
        after.add_record(changed)
        after.add_record(b)
        changes = before.changes(after, target)
        self.assertEquals(1, len(changes))
        update = changes[0]
        self.assertIsInstance(update, Update)
github github / octodns / tests / test_octodns_provider_base.py View on Github external
# MAX_SAFE_DELETE_PCENT is safe when more
        # than MIN_EXISTING_RECORDS exist
        zone = Zone('unit.tests.', [])
        record = Record.new(zone, 'a', {
            'ttl': 30,
            'type': 'A',
            'value': '1.2.3.4',
        })

        for i in range(int(Plan.MIN_EXISTING_RECORDS)):
            zone.add_record(Record.new(zone, str(i), {
                            'ttl': 60,
                            'type': 'A',
                            'value': '2.3.4.5'
                            }))
        changes = [Delete(record)
                   for i in range(int(Plan.MIN_EXISTING_RECORDS *
                                      Plan.MAX_SAFE_DELETE_PCENT))]

        Plan(zone, zone, changes).raise_if_unsafe()
github github / octodns / tests / test_octodns_provider_mythicbeasts.py View on Github external
            self.assertEquals(1, len(filter(lambda d: isinstance(d, Delete),
                              plan.changes)))