diff --git a/README.md b/README.md
index f395b1e7..6764bd7a 100644
--- a/README.md
+++ b/README.md
@@ -5,6 +5,7 @@ This is a tool to sync data from different sources to a NetBox instance.
 Available source types:
 * VMware vCenter Server
 * [bb-ricardo/check_redfish](https://github.com/bb-Ricardo/check_redfish) inventory files
+* [DigitalOcean](docs/source_digitalocean.md)
 
 **IMPORTANT: READ INSTRUCTIONS CAREFULLY BEFORE RUNNING THIS PROGRAM**
 
diff --git a/docs/source_digitalocean.md b/docs/source_digitalocean.md
new file mode 100644
index 00000000..fc0c2c64
--- /dev/null
+++ b/docs/source_digitalocean.md
@@ -0,0 +1,116 @@
+# DigitalOcean source
+
+The `digitalocean` source reads Droplets, attached block volumes, and managed
+database clusters from the DigitalOcean API and reconciles them into NetBox.
+API access is read-only; inventory changes are written to NetBox by the normal
+netbox-sync process.
+
+## Installation and access
+
+The source uses the `requests` dependency already installed by netbox-sync; no
+DigitalOcean SDK is required. Create a DigitalOcean API token with read access
+to Droplets and volumes. If managed database ingestion is enabled, the token
+must also be able to read databases.
+
+Keep the API token out of version control and provide it through a protected
+configuration file or the application's standard source environment variables.
+
+## Configuration
+
+Add a source section to the existing settings file:
+
+```ini
+[source/example-digitalocean]
+type = digitalocean
+api_token = REPLACE_WITH_READ_ONLY_API_TOKEN
+region_mapping = fra1=Germany, nyc1=United States
+sync_volumes = True
+sync_databases = True
+```
+
+`api_token` is required. `sync_volumes` and `sync_databases` default to `True`.
+Set either to `False` when the token should not query that API resource.
+Accounts without managed databases can leave `sync_databases` enabled:
+both `{"databases": []}` and `{"databases": null}` are treated as empty
+database collections. Missing collection fields and other malformed values
+remain errors rather than being treated as empty inventory.
+Generate the complete configuration with:
+
+```sh
+python netbox-sync.py -g
+```
+
+`region_mapping` is optional. It accepts comma-separated, exact
+`DigitalOcean region slug=NetBox Region name` pairs. Quote destination names
+containing commas, for example `nyc1="New York, USA"`. Duplicate source slugs,
+empty entries, and Region names longer than 100 characters are rejected.
+
+On NetBox 4.2+, the source reuses a Region with the destination name, then
+tries its generated slug, and creates it if neither exists. Explicit mappings
+take precedence over Regions matching the DigitalOcean region. Without a
+mapping, the DigitalOcean region name and slug are used (for example
+`Frankfurt 1` and `fra1`). When only a region slug is provided by DigitalOcean,
+it is also used as the name. Existing Regions are not renamed.
+
+Multiple DigitalOcean regions can map to the same NetBox Region; their
+virtualization clusters remain separate and retain their DigitalOcean names.
+Existing clusters in the source's cluster group are updated in place when
+their scope changes, including previously unscoped clusters. Ambiguous
+duplicate clusters in that group cause an explicit error instead of choosing
+one arbitrarily.
+
+The NetBox token needs permission to add Regions when they do not already
+exist. On NetBox versions before 4.2, Region creation and mapping are not
+applied; clusters remain unscoped.
+
+## NetBox mapping
+
+| DigitalOcean | NetBox |
+| --- | --- |
+| Region | Virtualization cluster scoped to a reused or automatically created NetBox Region on NetBox 4.2+ |
+| Droplet | Virtual machine |
+| Droplet size | VM vCPUs and memory |
+| Droplet image distribution/name | VM platform |
+| Droplet public/private IPv4 and IPv6 | VM interfaces and IP addresses |
+| Droplet root disk | VM disk capacity on older NetBox; virtual disk on NetBox 3.7+ |
+| Attached block volume | Additional virtual disk on NetBox 3.7+ |
+| DigitalOcean resource tags | NetBox tags |
+| Managed database cluster | VM inventory record with `db-` name prefix and metadata custom fields |
+
+DigitalOcean reports Droplet and volume disk sizes in gigabytes; these are
+converted to megabytes for NetBox using 1024 MB per reported GB. Database
+clusters are represented as one VM-like record per cluster because this
+project has no native managed-database object type. Database engine, version,
+size, node count, endpoint host, and port are stored in `digitalocean_*` custom
+fields. Connection credentials and connection URIs are never copied. Database
+endpoint hostnames are not resolved into IP addresses.
+
+DigitalOcean private and public network addresses are assigned to interfaces
+named after their network type. The source does not create NetBox prefixes or
+sites.
+
+Droplets with the same name in different clusters remain distinct. Interfaces
+and virtual disks are matched to their resolved parent VM, not just its name.
+If an earlier run moved child objects between same-named VMs, inspect the next
+dry run for interface, disk, and primary-IP changes before enabling writes.
+
+IP assignments can be repaired automatically when the existing IP belongs to
+the current source, either through this run's ownership or its persisted
+`Source: <source name>` tag. Conflicting source ownership or uncertain ownership
+stops the run before the normal NetBox write/prune phase instead of silently
+skipping a discovered IP. Two interfaces claiming the same IP in one source run
+also cause an explicit error. Source section names must remain stable for
+persisted ownership tags to match. If an earlier run already removed an IP's
+source tag or marked it orphaned, verify its owner and restore the correct tag
+or repair its assignment manually before retrying.
+
+## Validation
+
+Start with a dry run:
+
+```sh
+python netbox-sync.py -n -c settings.ini
+```
+
+Confirm the discovered Droplet, volume, and database cluster counts and inspect
+the resulting NetBox object changes before enabling writes or pruning.
diff --git a/module/config/file_output.py b/module/config/file_output.py
index 6a002f15..eac26936 100644
--- a/module/config/file_output.py
+++ b/module/config/file_output.py
@@ -17,6 +17,7 @@
 from module.sources.vmware.config import VMWareConfig
 from module.sources.check_redfish.config import CheckRedfishConfig
 from module.sources.hetzner.config import HetznerConfig
+from module.sources.digitalocean.config import DigitalOceanConfig
 from module.common.logging import get_logger
 from module.config import default_config_file_path, source_config_section_name
 from module.config.files import ConfigFile, ConfigFileINI, ConfigFileYAML
@@ -35,7 +36,8 @@ class ConfigFileOutput(DescriptionFormatterMixin):
     source_config_list = [
         VMWareConfig,
         CheckRedfishConfig,
-        HetznerConfig
+        HetznerConfig,
+        DigitalOceanConfig
     ]
 
     header = f"Welcome to the {__description__} configuration file."
diff --git a/module/netbox/inventory.py b/module/netbox/inventory.py
index af39be89..690c6b1e 100644
--- a/module/netbox/inventory.py
+++ b/module/netbox/inventory.py
@@ -128,6 +128,15 @@ def get_by_data(self, object_type, data=None):
 
         # try to find by primary/secondary key
         elif data.get(object_type.primary_key) is not None:
+            if object_type in (NBVMInterface, NBVirtualDisk) and isinstance(data.get("virtual_machine"), NBVM):
+                # VM names are only unique within a cluster; use the resolved parent object.
+                for this_object in self.get_all_items(object_type):
+                    if this_object.data.get("virtual_machine") is data["virtual_machine"] and \
+                            str(this_object.data.get(object_type.primary_key)).lower() == \
+                            str(data[object_type.primary_key]).lower():
+                        return this_object
+                return None
+
             object_name_to_find = None
             for this_object in self.get_all_items(object_type):
 
diff --git a/module/netbox/object_classes.py b/module/netbox/object_classes.py
index e6e8c813..cef8a36f 100644
--- a/module/netbox/object_classes.py
+++ b/module/netbox/object_classes.py
@@ -1470,7 +1470,6 @@ class NBRegion(NetBoxObject):
     object_type = "dcim.region"
     primary_key = "name"
     prune = False
-    read_only = True
 
     def __init__(self, *args, **kwargs):
         self.data_model = {
diff --git a/module/sources/__init__.py b/module/sources/__init__.py
index 0666da30..26e98154 100644
--- a/module/sources/__init__.py
+++ b/module/sources/__init__.py
@@ -10,6 +10,7 @@
 # define all available sources here
 from module.sources.vmware.connection import VMWareHandler
 from module.sources.hetzner.connection import HetznerHandler
+from module.sources.digitalocean.connection import DigitalOceanHandler
 from module.sources.check_redfish.import_inventory import CheckRedfish
 
 from module.common.logging import get_logger
@@ -19,7 +20,7 @@
 from module.config import source_config_section_name
 
 # list of valid sources
-valid_sources = [VMWareHandler, CheckRedfish, HetznerHandler]
+valid_sources = [VMWareHandler, CheckRedfish, HetznerHandler, DigitalOceanHandler]
 
 
 def validate_source(source_class_object=None, state="pre"):
diff --git a/module/sources/digitalocean/__init__.py b/module/sources/digitalocean/__init__.py
new file mode 100644
index 00000000..9591209d
--- /dev/null
+++ b/module/sources/digitalocean/__init__.py
@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+#  Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved.
+#
+#  This work is licensed under the terms of the MIT license.
+#  For a copy, see file LICENSE.txt included in this
+#  repository or visit: <https://opensource.org/licenses/MIT>.
+
+from module.sources.digitalocean.connection import DigitalOceanHandler
diff --git a/module/sources/digitalocean/client.py b/module/sources/digitalocean/client.py
new file mode 100644
index 00000000..fca41007
--- /dev/null
+++ b/module/sources/digitalocean/client.py
@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+#  Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved.
+#
+#  This work is licensed under the terms of the MIT license.
+#  For a copy, see file LICENSE.txt included in this
+#  repository or visit: <https://opensource.org/licenses/MIT>.
+
+from urllib.parse import urljoin, urlparse
+
+import requests
+
+
+class DigitalOceanClient:
+
+    api_url = "https://api.digitalocean.com/v2/"
+    timeout = 30
+
+    def __init__(self, token, session=None):
+        self.session = session or requests.Session()
+        self.headers = {
+            "Accept": "application/json",
+            "Authorization": f"Bearer {token}",
+            "User-Agent": "netbox-sync",
+        }
+
+    def get_collection(self, path, collection_name):
+        url = urljoin(self.api_url, path.lstrip("/"))
+        params = {"per_page": 200}
+        objects = []
+
+        while url is not None:
+            response = self.session.get(
+                url,
+                headers=self.headers,
+                params=params,
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            payload = response.json()
+            if not isinstance(payload, dict):
+                raise ValueError(f"DigitalOcean response for '{path}' is not a JSON object.")
+            collection = payload.get(collection_name)
+            # DigitalOcean can return {"databases": null} for accounts without databases.
+            if collection_name == "databases" and "databases" in payload and collection is None:
+                collection = []
+            if not isinstance(collection, list):
+                raise ValueError(f"DigitalOcean response for '{path}' has no '{collection_name}' list.")
+            objects.extend(collection)
+
+            links = payload.get("links") or {}
+            if not isinstance(links, dict):
+                raise ValueError(f"DigitalOcean response for '{path}' has invalid pagination links.")
+            pages = links.get("pages") or {}
+            if not isinstance(pages, dict):
+                raise ValueError(f"DigitalOcean response for '{path}' has invalid pagination pages.")
+            next_url = pages.get("next")
+            if next_url:
+                if not isinstance(next_url, str):
+                    raise ValueError(f"DigitalOcean response for '{path}' has a non-string pagination link.")
+                url = urljoin(url, next_url)
+                parsed_url = urlparse(url)
+                if parsed_url.scheme != "https" or parsed_url.hostname != "api.digitalocean.com" \
+                        or parsed_url.port not in (None, 443) or parsed_url.username is not None:
+                    raise ValueError("DigitalOcean API returned a pagination link outside its HTTPS API host.")
+            else:
+                url = None
+            params = None
+
+        return objects
+
+    def close(self):
+        self.session.close()
diff --git a/module/sources/digitalocean/config.py b/module/sources/digitalocean/config.py
new file mode 100644
index 00000000..96d975c1
--- /dev/null
+++ b/module/sources/digitalocean/config.py
@@ -0,0 +1,87 @@
+# -*- coding: utf-8 -*-
+#  Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved.
+#
+#  This work is licensed under the terms of the MIT license.
+#  For a copy, see file LICENSE.txt included in this
+#  repository or visit: <https://opensource.org/licenses/MIT>.
+
+from module.common.logging import get_logger
+from module.common.misc import quoted_split
+from module.config import source_config_section_name
+from module.config.base import ConfigBase
+from module.config.option import ConfigOption
+
+log = get_logger()
+
+
+class DigitalOceanConfig(ConfigBase):
+
+    section_name = source_config_section_name
+    source_name_example = "my-digitalocean-example"
+
+    def __init__(self):
+        self.options = [
+            ConfigOption(
+                "enabled",
+                bool,
+                default_value=True,
+                description="Enable or disable the DigitalOcean source."
+            ),
+            ConfigOption(
+                "type",
+                str,
+                default_value="digitalocean",
+                description="Source type identifier. Must remain 'digitalocean'."
+            ),
+            ConfigOption(
+                "api_token",
+                str,
+                mandatory=True,
+                sensitive=True,
+                description="DigitalOcean API token with read access to the resources being synced."
+            ),
+            ConfigOption(
+                "region_mapping",
+                str,
+                config_example="fra1=Germany, nyc1=United States",
+                description="Optional comma-separated DigitalOcean region slug=NetBox Region name mappings. "
+                            "Missing Regions are created on NetBox 4.2+. Unmapped regions use the "
+                            "DigitalOcean region name and slug."
+            ),
+            ConfigOption(
+                "sync_volumes",
+                bool,
+                default_value=True,
+                description="Sync attached block volumes as virtual disks."
+            ),
+            ConfigOption(
+                "sync_databases",
+                bool,
+                default_value=True,
+                description="Sync managed database clusters as VM inventory records."
+            ),
+        ]
+
+        super().__init__()
+
+    def validate_options(self):
+        option = self.get_option_by_name("region_mapping")
+        mapping = {}
+        if option.value is not None:
+            if not isinstance(option.value, str):
+                log.error("Config option 'region_mapping' must be a comma-separated string of slug=name mappings.")
+                self.set_validation_failed()
+                return
+
+            for entry in quoted_split(option.value):
+                slug, separator, name = entry.partition("=")
+                slug = slug.strip(' "\'')
+                name = name.strip(' "\'')
+                if not separator or not slug or not name or len(name) > 100 or slug in mapping:
+                    log.error(f"Invalid region_mapping entry '{entry}': expected a unique region slug and "
+                              "a non-empty NetBox Region name of at most 100 characters (slug=name).")
+                    self.set_validation_failed()
+                    continue
+                mapping[slug] = name
+
+        option.set_value(mapping)
diff --git a/module/sources/digitalocean/connection.py b/module/sources/digitalocean/connection.py
new file mode 100644
index 00000000..372fe3ab
--- /dev/null
+++ b/module/sources/digitalocean/connection.py
@@ -0,0 +1,432 @@
+# -*- coding: utf-8 -*-
+#  Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved.
+#
+#  This work is licensed under the terms of the MIT license.
+#  For a copy, see file LICENSE.txt included in this
+#  repository or visit: <https://opensource.org/licenses/MIT>.
+
+from ipaddress import ip_interface
+from packaging import version
+
+from module.common.logging import get_logger
+from module.netbox import (
+    NBCustomField, NBCluster, NBClusterGroup, NBClusterType, NBIPAddress, NBPlatform,
+    NBRegion, NBTag, NBVM, NBVMInterface, NBVirtualDisk, NetBoxObject,
+)
+from module.netbox.inventory import NetBoxInventory
+from module.sources.common.source_base import SourceBase
+from module.sources.digitalocean.client import DigitalOceanClient
+from module.sources.digitalocean.config import DigitalOceanConfig
+
+log = get_logger()
+
+
+class DigitalOceanHandler(SourceBase):
+
+    source_type = "digitalocean"
+    settings = DigitalOceanConfig()
+
+    dependent_netbox_objects = [
+        NBCustomField, NBCluster, NBClusterGroup, NBClusterType, NBIPAddress, NBPlatform,
+        NBRegion, NBTag, NBVM, NBVMInterface, NBVirtualDisk,
+    ]
+
+    def __init__(self, name=None):
+        if not name:
+            raise ValueError(f"Invalid value for attribute 'name': '{name}'.")
+
+        self.inventory = NetBoxInventory()
+        self.name = name
+        self.log = get_logger()
+        self._ip_claims = {}
+        self.set_source_tag()
+
+        settings_handler = DigitalOceanConfig()
+        settings_handler.source_name = name
+        self.settings = settings_handler.parse()
+
+        if self.settings.enabled is False:
+            log.info(f"Source '{name}' is currently disabled. Skipping")
+            return
+
+        self.init_successful = True
+
+    def apply(self):
+        self._ip_claims = {}
+        client = DigitalOceanClient(self.settings.api_token)
+
+        # Fetch everything before touching inventory so API failures cannot produce
+        # a partial run that later looks like missing resources.
+        try:
+            droplets = client.get_collection("droplets", "droplets")
+            volumes = client.get_collection("volumes", "volumes") if self.settings.sync_volumes else []
+            databases = client.get_collection("databases", "databases") if self.settings.sync_databases else []
+        finally:
+            client.close()
+
+        self.log.info(
+            f"Connected to DigitalOcean, found {len(droplets)} droplets, "
+            f"{len(volumes)} volumes and {len(databases)} database clusters"
+        )
+
+        cluster_group = self.inventory.add_update_object(
+            NBClusterGroup,
+            data={
+                "name": f"DigitalOcean: {self.name}",
+                "slug": NetBoxObject.format_slug(f"digitalocean-{self.name}"),
+            },
+            source=self,
+        )
+        cluster_type = self.inventory.add_update_object(
+            NBClusterType,
+            data={"name": "DigitalOcean", "slug": "digitalocean"},
+            source=self,
+        )
+        volumes_by_id = {str(volume["id"]): volume for volume in volumes}
+        clusters = {}
+
+        for droplet in droplets:
+            cluster = self._get_cluster(droplet.get("region"), cluster_group, cluster_type, clusters)
+            if cluster is None:
+                log.warning(f"Droplet '{droplet.get('name')}' has no region; skipping.")
+                continue
+            vm = self._sync_droplet(droplet, cluster)
+            self._sync_droplet_network(vm, droplet)
+            self._sync_droplet_disks(vm, droplet, volumes_by_id)
+
+        for database in databases:
+            cluster = self._get_cluster(database.get("region"), cluster_group, cluster_type, clusters)
+            if cluster is None:
+                log.warning(f"Database cluster '{database.get('name')}' has no region; skipping.")
+                continue
+            self._sync_database(database, cluster)
+
+    def _get_cluster(self, region, cluster_group, cluster_type, clusters):
+        if isinstance(region, dict):
+            region_slug = region.get("slug") or region.get("name")
+            region_name = region.get("name") or region_slug
+        else:
+            region_slug = region
+            region_name = region
+        if not region_slug:
+            return None
+
+        key = str(region_slug)
+        if key in clusters:
+            return clusters[key]
+
+        cluster_data = {
+            "name": f"{self.name}: {region_name}"[:100],
+            "type": cluster_type,
+            "group": cluster_group,
+        }
+        if version.parse(self.inventory.netbox_api_version) >= version.parse("4.2.0"):
+            mapped_name = self.settings.region_mapping.get(key)
+            target_name = mapped_name or region_name
+            target_slug = NetBoxObject.format_slug(mapped_name) if mapped_name else key
+            regions = self.inventory.get_all_items(NBRegion)
+            netbox_region = next((
+                item for item in regions if item.data.get("name") == target_name
+            ), None)
+            if netbox_region is None:
+                netbox_region = next((
+                    item for item in regions
+                    if item.data.get("slug") == target_slug or
+                    (mapped_name is None and item.data.get("name") == key)
+                ), None)
+            if netbox_region is None:
+                netbox_region = self.inventory.add_object(
+                    NBRegion, data={"name": target_name, "slug": target_slug}, source=self,
+                )
+            cluster_data["scope_type"] = NBRegion.object_type
+            cluster_data["scope_id"] = netbox_region
+
+        # Scope changes must update this source's cluster rather than create a duplicate.
+        candidates = [
+            item for item in self.inventory.get_all_items(NBCluster)
+            if item.data.get("name") == cluster_data["name"] and item.data.get("group") is cluster_group
+        ]
+        if len(candidates) > 1:
+            message = f"Multiple DigitalOcean clusters match '{cluster_data['name']}' in the source cluster group."
+            log.error(message)
+            raise ValueError(message)
+        if candidates:
+            cluster = candidates[0]
+            cluster.update(data=cluster_data, source=self)
+        else:
+            cluster = self.inventory.add_update_object(NBCluster, data=cluster_data, source=self)
+        clusters[key] = cluster
+        return cluster
+
+    def _sync_droplet(self, droplet, cluster):
+        name = str(droplet.get("name", "")).strip()
+        if not name:
+            log.warning(f"Skipping DigitalOcean droplet without a name (ID: {droplet.get('id')}).")
+            return None
+
+        size = droplet.get("size") or {}
+        memory = droplet.get("memory")
+        if memory is None:
+            memory = size.get("memory")
+        vcpus = droplet.get("vcpus")
+        if vcpus is None:
+            vcpus = size.get("vcpus")
+        disk_gb = droplet.get("disk")
+        if disk_gb is None:
+            disk_gb = size.get("disk")
+        vm_data = {
+            "name": name[:64],
+            "cluster": cluster,
+            "status": "active" if droplet.get("status") == "active" else "offline",
+            "serial": str(droplet.get("id")) if droplet.get("id") is not None else None,
+            "vcpus": vcpus,
+            "memory": memory,
+        }
+        if disk_gb is not None and version.parse(self.inventory.netbox_api_version) < version.parse("3.7.0"):
+            vm_data["disk"] = int(disk_gb) * 1024
+
+        platform = self._droplet_platform(droplet.get("image"))
+        if platform:
+            platform_name = platform[:100]
+            vm_data["platform"] = self.inventory.add_update_object(
+                NBPlatform,
+                data={"name": platform_name, "slug": NetBoxObject.format_slug(platform_name)},
+                source=self,
+            )
+        tags = [tag[:100] for tag in (droplet.get("tags") or []) if isinstance(tag, str) and tag]
+        if tags:
+            self._ensure_tags(tags)
+            vm_data["tags"] = tags
+
+        custom_fields = self._custom_fields({
+            "digitalocean_resource_type": "droplet",
+            "digitalocean_resource_id": droplet.get("id"),
+        })
+        if custom_fields:
+            vm_data["custom_fields"] = custom_fields
+
+        return self.inventory.add_update_object(NBVM, data=vm_data, source=self)
+
+    @staticmethod
+    def _droplet_platform(image):
+        if not isinstance(image, dict):
+            return None
+        distribution = image.get("distribution")
+        image_name = image.get("name")
+        if image_name and distribution and distribution.lower() not in image_name.lower():
+            return f"{distribution} {image_name}"
+        return image_name or distribution
+
+    def _sync_droplet_network(self, vm, droplet):
+        if vm is None:
+            return
+
+        networks = droplet.get("networks") or {}
+        primary = {4: None, 6: None}
+        primary_is_public = {4: False, 6: False}
+        for family in ("v4", "v6"):
+            for address in networks.get(family, []) or []:
+                ip_value = address.get("ip_address")
+                if ip_value is None:
+                    continue
+                ip_value = str(ip_value)
+                interface_name = str(address.get("type") or "network").lower()
+                try:
+                    ip_obj = self._parse_network_address(ip_value, address)
+                except ValueError:
+                    log.warning(f"Invalid IP address '{ip_value}' on DigitalOcean droplet '{vm.data['name']}'.")
+                    continue
+
+                interface = self.inventory.add_update_object(
+                    NBVMInterface,
+                    data={
+                        "name": interface_name[:64],
+                        "virtual_machine": vm,
+                        "enabled": True,
+                    },
+                    source=self,
+                )
+                ip_object = self._sync_ip(ip_obj, interface)
+                if ip_object is None:
+                    continue
+                ip_version = ip_obj.version
+                is_public = interface_name == "public"
+                if primary[ip_version] is None or (is_public and not primary_is_public[ip_version]):
+                    primary[ip_version] = ip_object
+                    primary_is_public[ip_version] = is_public
+
+        primary_data = {}
+        if primary[4] is not None:
+            primary_data["primary_ip4"] = primary[4]
+        if primary[6] is not None:
+            primary_data["primary_ip6"] = primary[6]
+        if primary_data:
+            vm.update(data=primary_data, source=self)
+
+    @staticmethod
+    def _parse_network_address(ip_value, address):
+        prefix = address.get("netmask")
+        if prefix is None:
+            prefix = address.get("cidr")
+        if prefix is None:
+            prefix = 32 if ":" not in ip_value else 128
+        return ip_interface(f"{ip_value}/{prefix}")
+
+    def _sync_ip(self, ip_obj, interface):
+        claim_key = str(ip_obj.ip)
+        previous_claim = self._ip_claims.get(claim_key)
+        if previous_claim is not None and previous_claim is not interface:
+            message = (f"DigitalOcean source '{self.name}' reports IP '{ip_obj}' on multiple interfaces. "
+                       f"Skipping assignment to '{interface.get_display_name()}'; keeping the first assignment.")
+            log.error(message)
+            return None
+
+        ip_data = {
+            "address": str(ip_obj),
+            "assigned_object_type": "virtualization.vminterface",
+            "assigned_object_id": interface,
+        }
+        existing = self.inventory.get_by_data(NBIPAddress, data={"address": str(ip_obj)})
+        if existing is not None:
+            assigned_object = existing.data.get("assigned_object_id")
+            if assigned_object is not None and assigned_object is not interface:
+                source_tags = {tag for tag in existing.get_tags() if tag.startswith("Source: ")}
+                same_source = existing.source is self or self.source_tag in source_tags
+                other_source = (
+                    (existing.source is not None and existing.source is not self) or
+                    bool(source_tags - {self.source_tag})
+                )
+                if not same_source or other_source:
+                    message = (
+                        f"DigitalOcean IP '{ip_obj}' has conflicting or uncertain source ownership; "
+                        f"cannot reassign it for source '{self.name}'. "
+                        f"Skipping assignment to '{interface.get_display_name()}'; resolve the existing assignment."
+                    )
+                    log.error(message)
+                    if existing.source is None and self.source_tag in source_tags:
+                        # Retain previously managed IPs without changing their assignment or ownership tags.
+                        existing.set_source(self)
+                    return None
+                log.info(f"Repairing DigitalOcean IP '{ip_obj}' assignment for source '{self.name}' "
+                         f"to interface '{interface.get_display_name()}'.")
+        result = self.inventory.add_update_object(NBIPAddress, data=ip_data, source=self)
+        self._ip_claims[claim_key] = interface
+        return result
+
+    def _sync_droplet_disks(self, vm, droplet, volumes_by_id):
+        if vm is None:
+            return
+
+        disk_gb = droplet.get("disk")
+        if disk_gb is None:
+            disk_gb = (droplet.get("size") or {}).get("disk")
+        total_disk_mb = int(disk_gb or 0) * 1024
+        attached_volumes = [
+            volumes_by_id[str(volume_id)]
+            for volume_id in (droplet.get("volume_ids") or [])
+            if str(volume_id) in volumes_by_id
+        ]
+        total_disk_mb += sum(int(volume.get("size_gigabytes", 0)) * 1024 for volume in attached_volumes)
+
+        if version.parse(self.inventory.netbox_api_version) < version.parse("3.7.0"):
+            vm.update(data={"disk": total_disk_mb}, source=self)
+            return
+
+        if disk_gb is not None:
+            self._sync_virtual_disk(
+                vm, "root", int(disk_gb) * 1024, "DigitalOcean Droplet root disk"
+            )
+        for volume in attached_volumes:
+            self._sync_virtual_disk(
+                vm,
+                f"do-{volume['id']}"[:64],
+                int(volume.get("size_gigabytes", 0)) * 1024,
+                volume.get("name"),
+            )
+
+    def _sync_database(self, database, cluster):
+        name = str(database.get("name", "")).strip()
+        if not name:
+            log.warning(f"Skipping DigitalOcean database cluster without a name (ID: {database.get('id')}).")
+            return
+
+        vm_data = {
+            "name": f"db-{name}"[:64],
+            "cluster": cluster,
+            "status": "active" if database.get("status") == "online" else "offline",
+            "serial": str(database.get("id")) if database.get("id") is not None else None,
+            "comments": "DigitalOcean managed database cluster",
+        }
+        engine = database.get("engine")
+        db_version = database.get("version")
+        platform = " ".join(str(value) for value in (engine, db_version) if value)
+        if platform:
+            platform_name = platform[:100]
+            vm_data["platform"] = self.inventory.add_update_object(
+                NBPlatform,
+                data={"name": platform_name, "slug": NetBoxObject.format_slug(platform_name)},
+                source=self,
+            )
+        tags = [tag[:100] for tag in (database.get("tags") or []) if isinstance(tag, str) and tag]
+        if tags:
+            self._ensure_tags(tags)
+            vm_data["tags"] = tags
+
+        connection = database.get("private_connection") or database.get("connection") or {}
+        custom_fields = self._custom_fields({
+            "digitalocean_resource_type": "database",
+            "digitalocean_resource_id": database.get("id"),
+            "digitalocean_engine": engine,
+            "digitalocean_version": db_version,
+            "digitalocean_size": database.get("size"),
+            "digitalocean_node_count": database.get("num_nodes"),
+            "digitalocean_endpoint": connection.get("host"),
+            "digitalocean_port": connection.get("port"),
+        })
+        if custom_fields:
+            vm_data["custom_fields"] = custom_fields
+        vm = self.inventory.add_update_object(NBVM, data=vm_data, source=self)
+
+        storage_size = database.get("storage_size_mib")
+        if storage_size is not None:
+            if version.parse(self.inventory.netbox_api_version) < version.parse("3.7.0"):
+                vm.update(data={"disk": int(storage_size)}, source=self)
+            else:
+                self._sync_virtual_disk(vm, "database-storage", int(storage_size), "Managed database storage")
+
+    def _custom_fields(self, values):
+        custom_fields = {}
+        for field_name, value in values.items():
+            if value is None:
+                continue
+            self.add_update_custom_field({
+                "name": field_name,
+                "label": field_name.replace("digitalocean_", "DO ").replace("_", " ")[:50],
+                "object_types": [NBVM.object_type],
+                "type": "text",
+                "description": f"DigitalOcean metadata for source '{self.name}'",
+            })
+            custom_fields[field_name] = str(value)
+        return custom_fields
+
+    def _ensure_tags(self, tags):
+        for tag in tags:
+            if not isinstance(tag, str) or not tag:
+                continue
+            self.inventory.add_update_object(
+                NBTag,
+                data={"name": tag, "slug": NetBoxObject.format_slug(tag)},
+                source=self,
+            )
+
+    def _sync_virtual_disk(self, vm, name, size_mb, description=None):
+        self.inventory.add_update_object(
+            NBVirtualDisk,
+            data={
+                "name": name,
+                "virtual_machine": vm,
+                "size": size_mb,
+                "description": description,
+            },
+            source=self,
+        )
diff --git a/tests/test_digitalocean_source.py b/tests/test_digitalocean_source.py
new file mode 100644
index 00000000..12681b07
--- /dev/null
+++ b/tests/test_digitalocean_source.py
@@ -0,0 +1,581 @@
+from ipaddress import ip_interface
+from unittest.mock import Mock
+
+import pytest
+
+from module.config.file_output import ConfigFileOutput
+from module.config.files import ConfigFileINI, ConfigFileYAML
+from module.config.parser import ConfigParser
+from module.netbox import (
+    NBCluster, NBClusterGroup, NBCustomField, NBIPAddress, NBPlatform, NBRegion, NBTag, NBVM, NBVMInterface,
+    NBVirtualDisk,
+)
+from module.netbox.connection import NetBoxHandler
+from module.sources import valid_sources
+from module.sources.digitalocean.config import DigitalOceanConfig
+from module.sources.digitalocean.connection import DigitalOceanHandler
+from module.sources.digitalocean.client import DigitalOceanClient
+
+
+@pytest.fixture
+def digitalocean(inventory, monkeypatch):
+    parser = ConfigParser()
+    monkeypatch.setattr(parser, "content", {
+        "source": {"cloud": {"type": "digitalocean", "api_token": "test-token"}}
+    })
+    monkeypatch.setattr(parser, "parsing_finished", True)
+    client = Mock()
+    client.get_collection.side_effect = lambda path, key: {
+        "droplets": [dict(
+            id=123,
+            name="web-1",
+            status="active",
+            region={"slug": "nyc1", "name": "New York 1"},
+            memory=2048,
+            vcpus=2,
+            disk=25,
+            image={"distribution": "Ubuntu", "name": "Ubuntu 24.04"},
+            tags=["production", "web"],
+            volume_ids=["volume-1"],
+            networks={
+                "v4": [
+                    {"ip_address": "10.10.0.5", "netmask": "255.255.255.0", "type": "private"},
+                    {"ip_address": "203.0.113.10", "netmask": "255.255.255.0", "type": "public"},
+                ],
+                "v6": [
+                    {"ip_address": "2001:db8::5", "cidr": 64, "type": "public"},
+                ],
+            },
+        )],
+        "volumes": [dict(
+            id="volume-1",
+            name="database-data",
+            size_gigabytes=20,
+            droplet_ids=[123],
+        )],
+        "databases": [dict(
+            id="database-uuid",
+            name="orders",
+            region={"slug": "nyc1", "name": "New York 1"},
+            status="online",
+            engine="pg",
+            version="16",
+            size="db-s-2vcpu-4gb",
+            num_nodes=2,
+            storage_size_mib=102400,
+            private_connection={
+                "host": "db-private.example.org",
+                "port": 25060,
+                "uri": "postgres://user:secret@db-private.example.org:25060/db",
+            },
+            connection={
+                "host": "db-public.example.org",
+                "port": 25060,
+                "password": "do-not-copy",
+            },
+            tags=["production"],
+        )],
+    }[path]
+    monkeypatch.setattr(
+        "module.sources.digitalocean.connection.DigitalOceanClient",
+        Mock(return_value=client),
+    )
+    return inventory, client
+
+
+def test_source_registration_and_settings_generation(monkeypatch):
+    assert DigitalOceanHandler in valid_sources
+    assert DigitalOceanConfig in ConfigFileOutput.source_config_list
+    assert DigitalOceanHandler.implements("digitalocean")
+
+    for file_type in (ConfigFileINI, ConfigFileYAML):
+        output = ConfigFileOutput.__new__(ConfigFileOutput)
+        output.config_file_type = file_type
+        output.comment_prefix = file_type.comment_prefix
+        output.lines = []
+        output.format()
+        text = "\n".join(output.lines)
+        assert "my-digitalocean-example" in text
+        assert "sync_databases" in text
+        assert "api_token" in text
+        assert "region_mapping" in text
+
+
+@pytest.mark.parametrize("mapping, expected", [
+    (None, {}),
+    ("fra1=Germany, nyc1=United States", {"fra1": "Germany", "nyc1": "United States"}),
+    (' fra1 = "Germany", nyc1 = "New York, USA" ', {"fra1": "Germany", "nyc1": "New York, USA"}),
+])
+def test_region_mapping_configuration(mapping, expected):
+    config = DigitalOceanConfig()
+    config.source_name = "cloud"
+    config.config_content = {"source": {"cloud": {"api_token": "test-token", "region_mapping": mapping}}}
+
+    assert config.parse(do_log=False).region_mapping == expected
+
+
+@pytest.mark.parametrize("mapping", [
+    "fra1", "=Germany", "fra1=", "fra1=Germany,",
+    "fra1=Germany,fra1=Europe", ["fra1=Germany"], "fra1=" + "x" * 101,
+])
+def test_invalid_region_mapping_is_rejected(mapping, caplog):
+    config = DigitalOceanConfig()
+    config.source_name = "cloud"
+    config.config_content = {"source": {"cloud": {"api_token": "test-token", "region_mapping": mapping}}}
+
+    with pytest.raises(SystemExit):
+        config.parse(do_log=False)
+
+    assert "region_mapping" in caplog.text
+
+
+@pytest.mark.parametrize("region, name, slug", [
+    ({"slug": "fra1", "name": "Frankfurt 1"}, "Frankfurt 1", "fra1"),
+    ("fra1", "fra1", "fra1"),
+])
+def test_missing_region_is_created_and_reused(digitalocean, region, name, slug, caplog):
+    inventory, _ = digitalocean
+    source = DigitalOceanHandler("cloud")
+
+    cluster = source._get_cluster(region, None, None, {})
+    netbox_region = cluster.data["scope_id"]
+    assert netbox_region.data["name"] == name
+    assert netbox_region.data["slug"] == slug
+    assert netbox_region.read_only is False
+    assert {"name", "slug"} <= set(netbox_region.updated_items)
+    assert cluster.data["scope_type"] == NBRegion.object_type
+    assert source._get_cluster(region, None, None, {}) is cluster
+    assert len(inventory.get_all_items(NBRegion)) == 1
+    assert "Unable to determine second key" not in caplog.text
+
+
+@pytest.mark.parametrize("existing_name, existing_slug", [
+    ("Germany", "de"), ("Federal Republic", "germany"), (None, None),
+])
+def test_region_mapping_overrides_provider_match(digitalocean, existing_name, existing_slug):
+    inventory, _ = digitalocean
+    provider_region = inventory.add_object(
+        NBRegion, data={"name": "Frankfurt 1", "slug": "fra1"}, read_from_netbox=True,
+    )
+    existing = None
+    if existing_name is not None:
+        existing = inventory.add_object(
+            NBRegion, data={"name": existing_name, "slug": existing_slug}, read_from_netbox=True,
+        )
+    source = DigitalOceanHandler("cloud")
+    source.settings.region_mapping = {"fra1": "Germany", "fra2": "Germany"}
+    cluster = source._get_cluster({"slug": "fra1", "name": "Frankfurt 1"}, None, None, {})
+
+    mapped_region = cluster.data["scope_id"]
+    assert mapped_region is not provider_region
+    if existing is not None:
+        assert mapped_region is existing
+        assert mapped_region.data["name"] == existing_name
+        assert mapped_region.data["slug"] == existing_slug
+        assert not mapped_region.updated_items
+    else:
+        assert mapped_region.data["name"] == "Germany"
+        assert mapped_region.data["slug"] == "germany"
+    other_cluster = source._get_cluster({"slug": "fra2", "name": "Frankfurt 2"}, None, None, {})
+    assert other_cluster.data["scope_id"] is mapped_region
+    assert len(inventory.get_all_items(NBRegion)) == 2
+
+
+@pytest.mark.parametrize("previously_unscoped", [True, False])
+def test_existing_source_cluster_is_rescoped_without_duplicates(digitalocean, previously_unscoped):
+    inventory, _ = digitalocean
+    source = DigitalOceanHandler("cloud")
+    source.apply()
+    cluster = inventory.get_all_items(NBCluster)[0]
+    original_vm = inventory.get_all_items(NBVM)[0]
+    if previously_unscoped:
+        cluster.data.pop("scope_id")
+        cluster.data.pop("scope_type")
+
+    source.settings.region_mapping = {"nyc1": "United States"}
+    source.apply()
+
+    assert inventory.get_all_items(NBCluster) == [cluster]
+    assert original_vm.data["cluster"] is cluster
+    assert cluster.data["scope_id"].data["name"] == "United States"
+
+
+def test_old_netbox_does_not_create_region_scopes(digitalocean):
+    inventory, _ = digitalocean
+    inventory.netbox_api_version = "4.1.11"
+    source = DigitalOceanHandler("cloud")
+    source.settings.region_mapping = {"fra1": "Germany"}
+
+    cluster = source._get_cluster({"slug": "fra1", "name": "Frankfurt 1"}, None, None, {})
+
+    assert "scope_id" not in cluster.data
+    assert "scope_type" not in cluster.data
+    assert not inventory.get_all_items(NBRegion)
+
+
+@pytest.mark.parametrize("name, slug", [
+    ("Frankfurt 1", "frankfurt"), ("Existing Frankfurt", "fra1"), ("fra1", "existing"),
+])
+def test_unmapped_region_reuses_existing_name_or_slug(digitalocean, name, slug):
+    inventory, _ = digitalocean
+    existing = inventory.add_object(
+        NBRegion, data={"name": name, "slug": slug}, read_from_netbox=True,
+    )
+    source = DigitalOceanHandler("cloud")
+
+    cluster = source._get_cluster({"slug": "fra1", "name": "Frankfurt 1"}, None, None, {})
+
+    assert cluster.data["scope_id"] is existing
+    assert existing.data["name"] == name
+    assert existing.data["slug"] == slug
+    assert not existing.updated_items
+    assert len(inventory.get_all_items(NBRegion)) == 1
+
+
+def test_region_is_written_before_cluster_with_integer_scope_id(digitalocean):
+    inventory, _ = digitalocean
+    source = DigitalOceanHandler("cloud")
+    cluster = source._get_cluster({"slug": "fra1", "name": "Frankfurt 1"}, None, None, {})
+    region = cluster.data["scope_id"]
+    handler = NetBoxHandler.__new__(NetBoxHandler)
+    handler.inventory = inventory
+    handler.request = Mock(side_effect=lambda object_type, req_type, data, nb_id: {
+        **data, "id": 101 if object_type is NBRegion else 201,
+    })
+
+    handler.update_instance()
+
+    calls = handler.request.call_args_list
+    assert [call.args[0] for call in calls] == [NBRegion, NBCluster]
+    assert all(call.kwargs["req_type"] == "POST" for call in calls)
+    assert calls[0].kwargs["data"]["name"] == "Frankfurt 1"
+    assert calls[1].kwargs["data"]["scope_id"] == 101
+    assert calls[1].kwargs["data"]["scope_type"] == NBRegion.object_type
+    assert cluster.data["scope_id"] is region
+    assert not region.is_new
+    assert not cluster.is_new
+    assert not region.updated_items
+    assert not cluster.updated_items
+    handler.request.reset_mock()
+    handler.update_instance()
+    handler.request.assert_not_called()
+
+
+def test_ambiguous_source_clusters_are_rejected(digitalocean, caplog):
+    inventory, _ = digitalocean
+    source = DigitalOceanHandler("cloud")
+    source.apply()
+    cluster = inventory.get_all_items(NBCluster)[0]
+    inventory.add_object(NBCluster, data=dict(cluster.data), read_from_netbox=True)
+
+    with pytest.raises(ValueError, match="Multiple DigitalOcean clusters"):
+        source.apply()
+
+    assert "Multiple DigitalOcean clusters" in caplog.text
+    assert len(inventory.get_all_items(NBCluster)) == 2
+
+
+def test_digitalocean_discovery_maps_droplets_volumes_and_database_clusters(digitalocean):
+    inventory, client = digitalocean
+    inventory.add_object(
+        NBRegion,
+        data={"name": "New York 1", "slug": "nyc1"},
+        read_from_netbox=True,
+    )
+    source = DigitalOceanHandler("cloud")
+    assert source.init_successful
+    inventory.resolve_relations()
+    source.apply()
+
+    client.close.assert_called_once()
+    assert [call.args for call in client.get_collection.call_args_list] == [
+        ("droplets", "droplets"),
+        ("volumes", "volumes"),
+        ("databases", "databases"),
+    ]
+    assert len(inventory.get_all_items(NBCluster)) == 1
+    assert len(inventory.get_all_items(NBClusterGroup)) == 1
+    cluster = inventory.get_all_items(NBCluster)[0]
+    assert cluster.data["scope_type"] == NBRegion.object_type
+    assert cluster.data["scope_id"].data["slug"] == "nyc1"
+    assert len(inventory.get_all_items(NBVM)) == 2
+    assert len(inventory.get_all_items(NBVMInterface)) == 2
+    assert len(inventory.get_all_items(NBIPAddress)) == 3
+    assert len(inventory.get_all_items(NBVirtualDisk)) == 3
+
+    droplet = next(vm for vm in inventory.get_all_items(NBVM) if vm.data["name"] == "web-1")
+    assert droplet.data["status"] == "active"
+    assert droplet.data["serial"] == "123"
+    assert droplet.data["vcpus"] == 2
+    assert droplet.data["memory"] == 2048
+    assert droplet.data["platform"].data["name"] == "Ubuntu 24.04"
+    assert droplet.data["primary_ip4"].data["address"] == "203.0.113.10/24"
+    assert droplet.data["primary_ip6"].data["address"] == "2001:db8::5/64"
+    assert {tag.data["name"] for tag in droplet.data["tags"]} == {"production", "web"}
+    assert {
+        disk.data["name"]: disk.data["size"]
+        for disk in inventory.get_all_items(NBVirtualDisk)
+        if disk.data["virtual_machine"] is droplet
+    } == {"root": 25600, "do-volume-1": 20480}
+
+    database = next(vm for vm in inventory.get_all_items(NBVM) if vm.data["name"] == "db-orders")
+    assert database.data["status"] == "active"
+    assert database.data["serial"] == "database-uuid"
+    assert database.data["platform"].data["name"] == "pg 16"
+    assert database.data["custom_fields"]["digitalocean_resource_type"] == "database"
+    assert database.data["custom_fields"]["digitalocean_engine"] == "pg"
+    assert database.data["custom_fields"]["digitalocean_version"] == "16"
+    assert database.data["custom_fields"]["digitalocean_node_count"] == "2"
+    assert database.data["custom_fields"]["digitalocean_endpoint"] == "db-private.example.org"
+    assert database.data["custom_fields"]["digitalocean_port"] == "25060"
+    assert "secret" not in repr(database.data)
+    assert len(inventory.get_all_items(NBCustomField)) == 8
+    assert {tag.data["name"] for tag in inventory.get_all_items(NBTag)} == {
+        "production", "web",
+    }
+
+    source.apply()
+    assert len(inventory.get_all_items(NBCluster)) == 1
+    assert len(inventory.get_all_items(NBVM)) == 2
+    assert len(inventory.get_all_items(NBVirtualDisk)) == 3
+    assert len(inventory.get_all_items(NBPlatform)) == 2
+    assert len(inventory.get_all_items(NBClusterGroup)) == 1
+    assert len(inventory.get_all_items(NBTag)) == 2
+
+
+def test_same_named_droplets_in_different_clusters_keep_their_children(digitalocean):
+    inventory, client = digitalocean
+    droplets = [
+        {
+            "id": index,
+            "name": "prod-netops-FE-2",
+            "status": status,
+            "region": {"slug": slug, "name": region_name},
+            "disk": 25,
+            "networks": {"v4": [
+                {"ip_address": address, "cidr": 24, "type": "public"},
+            ]},
+        }
+        for index, slug, region_name, address, status in [
+            (1, "nyc3", "New York 3", "203.0.113.1", "off"),
+            (2, "nyc1", "New York 1", "203.0.113.2", "active"),
+        ]
+    ]
+    client.get_collection.side_effect = lambda path, key: droplets if path == "droplets" else []
+    source = DigitalOceanHandler("cloud")
+
+    for _ in range(2):
+        source.apply()
+        vms = inventory.get_all_items(NBVM)
+        assert len(vms) == 2
+        assert len(inventory.get_all_items(NBVMInterface)) == 2
+        assert len(inventory.get_all_items(NBVirtualDisk)) == 2
+        for vm in vms:
+            ip = vm.data["primary_ip4"]
+            interface = ip.data["assigned_object_id"]
+            assert interface.data["virtual_machine"] is vm
+            assert len(vm.get_virtual_disks()) == 1
+            assert vm.get_virtual_disks()[0].data["virtual_machine"] is vm
+
+
+@pytest.mark.parametrize("object_type", [NBVMInterface, NBVirtualDisk])
+def test_vm_children_match_resolved_parent_not_duplicate_name(inventory, object_type):
+    clusters = [
+        inventory.add_object(NBCluster, data={"name": name}, read_from_netbox=True)
+        for name in ("New York 3", "New York 1")
+    ]
+    vms = [
+        inventory.add_object(NBVM, data={"name": "same-name", "cluster": cluster}, read_from_netbox=True)
+        for cluster in clusters
+    ]
+    children = [
+        inventory.add_object(
+            object_type, data={"name": "public", "virtual_machine": vm}, read_from_netbox=True,
+        )
+        for vm in vms
+    ]
+
+    for vm, child in zip(vms, children):
+        assert inventory.add_update_object(
+            object_type, data={"name": "PUBLIC", "virtual_machine": vm},
+        ) is child
+        assert child.data["virtual_machine"] is vm
+        assert "virtual_machine" not in child.updated_items
+
+
+@pytest.mark.parametrize("ownership", ["same-tag", "same-run", "other-tag", "unknown", "mixed-tags", "other-source"])
+def test_ip_conflict_repairs_only_confirmed_same_source(digitalocean, ownership):
+    inventory, _ = digitalocean
+    source = DigitalOceanHandler("cloud")
+    vms = [
+        inventory.add_object(NBVM, data={"name": name}, read_from_netbox=True)
+        for name in ("old-vm", "new-vm")
+    ]
+    interfaces = [
+        inventory.add_object(
+            NBVMInterface, data={"name": "public", "virtual_machine": vm}, read_from_netbox=True,
+        )
+        for vm in vms
+    ]
+    ip = inventory.add_object(NBIPAddress, data={
+        "id": 19223, "address": "203.0.113.1/24",
+        "assigned_object_type": NBVMInterface.object_type, "assigned_object_id": interfaces[0],
+    }, read_from_netbox=True)
+    if ownership in ("same-tag", "mixed-tags", "other-source"):
+        ip.add_tags(source.source_tag)
+    if ownership in ("other-tag", "mixed-tags"):
+        ip.add_tags("Source: another-account")
+    if ownership == "same-run":
+        ip.source = source
+    if ownership == "other-source":
+        ip.source = Mock()
+    vms[0].data["primary_ip4"] = ip
+
+    if ownership in ("same-tag", "same-run"):
+        assert source._sync_ip(ip_interface("203.0.113.1/24"), interfaces[1]) is ip
+        assert ip.data["assigned_object_id"] is interfaces[1]
+        assert ip.source is source
+        assert "primary_ip4" in vms[0].unset_items
+    else:
+        with pytest.raises(ValueError, match="ownership"):
+            source._sync_ip(ip_interface("203.0.113.1/24"), interfaces[1])
+        assert ip.data["assigned_object_id"] is interfaces[0]
+
+
+def test_repaired_ip_is_not_orphaned(digitalocean):
+    inventory, _ = digitalocean
+    source = DigitalOceanHandler("cloud")
+    source.apply()
+    ip = inventory.get_by_data(NBIPAddress, data={"address": "203.0.113.10/24"})
+    target_interface = ip.data["assigned_object_id"]
+    wrong_vm = inventory.add_object(NBVM, data={"name": "wrong-vm"}, read_from_netbox=True)
+    wrong_interface = inventory.add_object(
+        NBVMInterface, data={"name": "public", "virtual_machine": wrong_vm}, read_from_netbox=True,
+    )
+    ip.data["assigned_object_id"] = wrong_interface
+    ip.source = None
+    ip.add_tags(["NetBox-synced", source.source_tag, "NetBox-synced: Orphaned"])
+    inventory.source_list = [source]
+
+    source.apply()
+    handler = Mock(
+        primary_tag="NetBox-synced", orphaned_tag="NetBox-synced: Orphaned",
+    )
+    inventory.tag_all_the_things(handler)
+
+    assert ip.data["assigned_object_id"] is target_interface
+    assert ip.source is source
+    assert source.source_tag in ip.get_tags()
+    assert "NetBox-synced: Orphaned" not in ip.get_tags()
+    assert target_interface.data["virtual_machine"].data["primary_ip4"] is ip
+
+
+@pytest.mark.parametrize("second_prefix", [24, 32])
+def test_conflicting_droplet_ip_claims_abort_before_writes(digitalocean, second_prefix):
+    inventory, client = digitalocean
+    droplets = [
+        {"id": index, "name": name, "region": "nyc1", "networks": {
+            "v4": [{"ip_address": "203.0.113.1", "cidr": prefix, "type": "public"}],
+        }}
+        for index, name, prefix in [(1, "first", 24), (2, "second", second_prefix)]
+    ]
+    client.get_collection.side_effect = lambda path, key: droplets if path == "droplets" else []
+    source = DigitalOceanHandler("cloud")
+
+    with pytest.raises(ValueError, match="multiple interfaces"):
+        source.apply()
+
+    ip = inventory.get_all_items(NBIPAddress)[0]
+    assert ip.data["assigned_object_id"].data["virtual_machine"].data["name"] == "first"
+
+
+def test_api_failure_closes_session_before_inventory_changes(digitalocean):
+    inventory, client = digitalocean
+    client.get_collection.side_effect = RuntimeError("API request failed")
+    source = DigitalOceanHandler("cloud")
+
+    with pytest.raises(RuntimeError, match="API request failed"):
+        source.apply()
+
+    client.close.assert_called_once()
+    assert not inventory.get_all_items(NBCluster)
+    assert not inventory.get_all_items(NBVM)
+
+
+def test_database_sync_can_be_disabled(digitalocean):
+    inventory, client = digitalocean
+    source = DigitalOceanHandler("cloud")
+    source.settings.sync_databases = False
+    source.settings.sync_volumes = False
+    source.apply()
+
+    assert [call.args for call in client.get_collection.call_args_list] == [("droplets", "droplets")]
+    assert all(vm.data["name"] != "db-orders" for vm in inventory.get_all_items(NBVM))
+
+
+def test_digitalocean_client_paginates_and_uses_bearer_auth():
+    response_one = Mock()
+    response_one.json.return_value = {
+        "droplets": [{"id": 1}],
+        "links": {"pages": {"next": "https://api.digitalocean.com/v2/droplets?page=2"}},
+    }
+    response_two = Mock()
+    response_two.json.return_value = {"droplets": [{"id": 2}], "links": {"pages": {}}}
+    session = Mock()
+    session.get.side_effect = [response_one, response_two]
+
+    client = DigitalOceanClient("test-token", session=session)
+
+    assert client.get_collection("droplets", "droplets") == [{"id": 1}, {"id": 2}]
+    first_call, second_call = session.get.call_args_list
+    assert first_call.kwargs["headers"]["Authorization"] == "Bearer test-token"
+    assert first_call.kwargs["params"] == {"per_page": 200}
+    assert second_call.kwargs["params"] is None
+
+
+@pytest.mark.parametrize("databases", [None, []])
+def test_digitalocean_client_accepts_empty_databases(databases):
+    response = Mock()
+    response.json.return_value = {"databases": databases}
+    session = Mock()
+    session.get.return_value = response
+    client = DigitalOceanClient("test-token", session=session)
+
+    assert client.get_collection("databases", "databases") == []
+    response.raise_for_status.assert_called_once()
+    session.get.assert_called_once()
+
+
+@pytest.mark.parametrize("path, payload", [
+    ("databases", {}),
+    ("databases", {"databases": {}}),
+    ("databases", {"databases": ""}),
+    ("databases", {"databases": False}),
+    ("droplets", {"droplets": None}),
+    ("volumes", {"volumes": None}),
+])
+def test_digitalocean_client_rejects_invalid_collections(path, payload):
+    response = Mock()
+    response.json.return_value = payload
+    session = Mock()
+    session.get.return_value = response
+    client = DigitalOceanClient("test-token", session=session)
+
+    with pytest.raises(ValueError, match=f"has no '{path}' list"):
+        client.get_collection(path, path)
+
+
+def test_digitalocean_client_rejects_external_pagination_link():
+    response = Mock()
+    response.json.return_value = {
+        "droplets": [],
+        "links": {"pages": {"next": "https://attacker.example/steal"}},
+    }
+    session = Mock()
+    session.get.return_value = response
+    client = DigitalOceanClient("test-token", session=session)
+
+    with pytest.raises(ValueError, match="pagination link"):
+        client.get_collection("droplets", "droplets")
+
+    session.get.assert_called_once()