zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-05-23 04:49:23 +02:00
Code Issues Projects Releases Packages Wiki Activity

some minor modifications

Browse source
This commit is contained in:
David Kennedy 2023-12-29 16:05:00 -05:00
parent d0ca1ccff1
commit 66bc63aad4
No known key found for this signature in database
GPG key ID: 6528A5386E66B96B
1 changed files with 43 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

82
src/registrar/models/domain.py
View file

@ -1608,7 +1608,8 @@ class Domain(TimeStampedModel, DomainHelper):
cache = self._extract_data_from_response(data_response) cache = self._extract_data_from_response(data_response)
cleaned = self._clean_cache(cache, data_response) cleaned = self._clean_cache(cache, data_response)
self._update_hosts_and_contacts(cleaned, fetch_hosts, fetch_contacts) self._update_hosts_and_contacts(cleaned, fetch_hosts, fetch_contacts)
self._update_hosts_and_ips_in_db(cleaned, fetch_hosts) if fetch_hosts:
self._update_hosts_and_ips_in_db(cleaned)
self._update_dates(cleaned) self._update_dates(cleaned)
self._cache = cleaned self._cache = cleaned
@ -1655,7 +1656,11 @@ class Domain(TimeStampedModel, DomainHelper):
return dnssec_data return dnssec_data
def _update_hosts_and_contacts(self, cleaned, fetch_hosts, fetch_contacts): def _update_hosts_and_contacts(self, cleaned, fetch_hosts, fetch_contacts):
"""Capture and cache old hosts and contacts from cache if they don't exist in cleaned""" """
Update hosts and contacts if fetch_hosts and/or fetch_contacts.
Additionally, capture and cache old hosts and contacts from cache if they
don't exist in cleaned
"""
old_cache_hosts = self._cache.get("hosts") old_cache_hosts = self._cache.get("hosts")
old_cache_contacts = self._cache.get("contacts") old_cache_contacts = self._cache.get("contacts")
@ -1670,50 +1675,49 @@ class Domain(TimeStampedModel, DomainHelper):
if old_cache_contacts is not None: if old_cache_contacts is not None:
cleaned["contacts"] = old_cache_contacts cleaned["contacts"] = old_cache_contacts
def _update_hosts_and_ips_in_db(self, cleaned, fetch_hosts): def _update_hosts_and_ips_in_db(self, cleaned):
"""Update hosts and host_ips in database if retrieved from registry. """Update hosts and host_ips in database if retrieved from registry.
Only called when fetch_hosts is True.
Parameters: Parameters:
self: the domain to be updated with hosts and ips from cleaned self: the domain to be updated with hosts and ips from cleaned
cleaned: dict containing hosts. Hosts are provided as a list of dicts, e.g. cleaned: dict containing hosts. Hosts are provided as a list of dicts, e.g.
[{"name": "ns1.example.com",}, {"name": "ns1.example.gov"}, "addrs": ["0.0.0.0"])] [{"name": "ns1.example.com",}, {"name": "ns1.example.gov"}, "addrs": ["0.0.0.0"])]
fetch_hosts: boolean indicating whether or not fetch_hosts was called
""" """
if fetch_hosts: cleaned_hosts = cleaned["hosts"]
cleaned_hosts = cleaned["hosts"] # Get all existing hosts from the database for this domain
# Get all existing hosts from the database for this domain existing_hosts_in_db = Host.objects.filter(domain=self)
existing_hosts_in_db = Host.objects.filter(domain=self) # Identify hosts to delete
# Identify hosts to delete cleaned_host_names = set(cleaned_host["name"] for cleaned_host in cleaned_hosts)
cleaned_host_names = set(cleaned_host["name"] for cleaned_host in cleaned_hosts) hosts_to_delete_from_db = [
hosts_to_delete_from_db = [ existing_host for existing_host in existing_hosts_in_db if existing_host.name not in cleaned_host_names
existing_host for existing_host in existing_hosts_in_db if existing_host.name not in cleaned_host_names ]
] # Delete hosts and their associated HostIP instances
# Delete hosts and their associated HostIP instances for host_to_delete in hosts_to_delete_from_db:
for host_to_delete in hosts_to_delete_from_db: # Delete associated HostIP instances
# Delete associated HostIP instances HostIP.objects.filter(host=host_to_delete).delete()
HostIP.objects.filter(host=host_to_delete).delete() # Delete the host itself
# Delete the host itself host_to_delete.delete()
host_to_delete.delete() # Update or create Hosts and HostIPs
# Update or create Hosts and HostIPs for cleaned_host in cleaned_hosts:
for cleaned_host in cleaned_hosts: # Check if the cleaned_host already exists
# Check if the cleaned_host already exists host_in_db, host_created = Host.objects.get_or_create(domain=self, name=cleaned_host["name"])
host_in_db, host_created = Host.objects.get_or_create(domain=self, name=cleaned_host["name"]) # Get cleaned list of ips for update
# Get cleaned list of ips for update cleaned_ips = cleaned_host["addrs"]
cleaned_ips = cleaned_host["addrs"] if not host_created:
if not host_created: # Get all existing ips from the database for this host
# Get all existing ips from the database for this host existing_ips_in_db = HostIP.objects.filter(host=host_in_db)
existing_ips_in_db = HostIP.objects.filter(host=host_in_db) # Identify IPs to delete
# Identify IPs to delete ips_to_delete_from_db = [
ips_to_delete_from_db = [ existing_ip for existing_ip in existing_ips_in_db if existing_ip.address not in cleaned_ips
existing_ip for existing_ip in existing_ips_in_db if existing_ip.address not in cleaned_ips ]
] # Delete IPs
# Delete IPs for ip_to_delete in ips_to_delete_from_db:
for ip_to_delete in ips_to_delete_from_db: # Delete the ip
# Delete the ip ip_to_delete.delete()
ip_to_delete.delete() # Update or create HostIP instances
# Update or create HostIP instances for ip_address in cleaned_ips:
for ip_address in cleaned_ips: HostIP.objects.get_or_create(address=ip_address, host=host_in_db)
HostIP.objects.get_or_create(address=ip_address, host=host_in_db)
def _update_dates(self, cleaned): def _update_dates(self, cleaned):
"""Update dates (expiration and creation) from cleaned""" """Update dates (expiration and creation) from cleaned"""