Fix latent provider bugs surfaced during the typing work

Follow-ups deferred from the add-typing PR (typed to conform / stopgapped,
not truly fixed):

- AzureVolume.source returned the raw source URI string but is declared
  Snapshot | MachineImage | None. Resolve the URI to the Snapshot object
  via snapshots.get(), mirroring the AWS/OpenStack implementations
  (returns None when there is no source). Adds a cross-provider regression
  test: a volume created from a snapshot now reports that snapshot as its
  source.

- AzureInstanceService._resolve_launch_options omitted the required
  `network` argument when merging multiple firewalls into a new one
  (pre-existing bug — the call would have raised TypeError). Create the
  merged firewall on the same network as the firewalls being combined.

- azure/helpers.py parse_url raised InvalidValueException with the template
  list in the `param` slot (misuse of `param: str`, stopgapped with str()).
  Name the offending parameter ('original_url') at both raise sites.

- Documented that GCPRouter.subnets returning the network's subnets is
  correct, not a bug: a GCP Cloud Router has no per-subnet attachment and
  automatically serves every subnet in its VPC network (unlike AWS route
  tables / OpenStack routers).

mypy 2.1 and flake8 clean. The Azure/GCP behaviour changes are validated by
the per-cloud integration suite in CI (not reproducible locally).

cloudbridge/providers/azure/helpers.py

@@ -53,7 +53,7 @@ def parse_url(template_urls: list[str], original_url: str) -> dict[str, str]:
     https://docs.microsoft.com/en-us/azure/virtual-machines/linux/cli-ps-findimage
     """
     if not original_url:
-        raise InvalidValueException(str(template_urls), original_url)
+        raise InvalidValueException('original_url', original_url)
     original_url_parts = original_url.split('/')
     if len(original_url_parts) == 1:
         original_url_parts = original_url.split(':')
@@ -64,7 +64,7 @@ def parse_url(template_urls: list[str], original_url: str) -> dict[str, str]:
         if len(template_url_parts) == len(original_url_parts):
             break
     if len(template_url_parts) != len(original_url_parts):
-        raise InvalidValueException(str(template_urls), original_url)
+        raise InvalidValueException('original_url', original_url)
     resource_param: dict[str, str] = {}
     for key, value in zip(template_url_parts, original_url_parts):
         if key.startswith('{') and key.endswith('}'):

cloudbridge/providers/azure/resources.py

@@ -455,7 +455,13 @@ class AzureVolume(BaseVolume):
 
     @property
     def source(self) -> Snapshot | MachineImage | None:
-        return self._volume.creation_data.source_uri
+        # ``source_uri`` is the resource URI of the disk's source (e.g. the
+        # snapshot it was copied from); resolve it to the Snapshot object the
+        # interface promises, mirroring the AWS/OpenStack implementations.
+        source_uri = self._volume.creation_data.source_uri
+        if source_uri:
+            return self._provider.storage.snapshots.get(source_uri)
+        return None
 
     @property
     def attachments(self) -> AttachmentInfo | None:

cloudbridge/providers/azure/services.py

@@ -858,23 +858,24 @@ class AzureInstanceService(BaseInstanceService):
 
         if isinstance(vm_firewalls, list) and len(vm_firewalls) > 0:
 
+            first_fw: Any
             if isinstance(vm_firewalls[0], VMFirewall):
+                first_fw = vm_firewalls[0]
                 vm_firewalls_ids = [cast(Any, fw).id for fw in vm_firewalls]
-                vm_firewall_id = cast(Any, vm_firewalls[0]).resource_id
             else:
                 vm_firewalls_ids = vm_firewalls
-                vm_firewall = self.provider.security.\
+                first_fw = self.provider.security.\
                     vm_firewalls.get(vm_firewalls[0])
-                vm_firewall_id = cast(Any, vm_firewall).resource_id
+            vm_firewall_id = first_fw.resource_id
 
             if len(vm_firewalls) > 1:
-                # FLAGGED FOR REVIEW: this create() omits the required
-                # ``network`` argument (pre-existing bug); cast to Any so the
-                # missing-arg is not masked by a fabricated value here.
-                new_fw = cast(Any, self.provider.security.vm_firewalls).\
-                    create(label='{0}-fw'.format(inst_name),
-                           description='Merge vm firewall {0}'.
-                           format(','.join(cast(Any, vm_firewalls_ids))))
+                # The merged firewall belongs to the same network as the
+                # firewalls being combined.
+                new_fw = self.provider.security.vm_firewalls.create(
+                    label='{0}-fw'.format(inst_name),
+                    network=first_fw.network_id,
+                    description='Merge vm firewall {0}'.format(
+                        ','.join(cast(Any, vm_firewalls_ids))))
 
                 for fw in vm_firewalls:
                     cast(Any, new_fw).add_rule(src_dest_fw=fw)

cloudbridge/providers/gcp/resources.py

@@ -1638,6 +1638,10 @@ class GCPRouter(BaseRouter):
 
     @property
     def subnets(self) -> Iterable[Subnet]:
+        # Unlike AWS route tables / OpenStack routers, a GCP Cloud Router has
+        # no per-subnet attachment: it automatically serves every subnet in
+        # its VPC network (see attach_subnet/detach_subnet). So the router's
+        # subnets are exactly the subnets of its network.
         network = cast(
             "GCPNetwork",
             self._provider.networking.networks.get(

tests/test_block_store_service.py

@@ -115,7 +115,6 @@ class CloudBlockStoreServiceTestCase(ProviderTestBase):
                     "Volume.description must be None or a string. Got: %s"
                     % test_vol.description)
                 self.assertIsNone(test_vol.source)
-                self.assertIsNone(test_vol.source)
                 self.assertIsNotNone(test_vol.create_time)
                 self.assertIsNotNone(test_vol.zone_id)
                 self.assertIsNone(test_vol.attachments)
@@ -229,6 +228,17 @@ class CloudBlockStoreServiceTestCase(ProviderTestBase):
                     sv_label, 1, snapshot=test_snap)
                 with cb_helpers.cleanup_action(lambda: snap_vol.delete()):
                     snap_vol.wait_till_ready()
+                    # A volume created from a snapshot should report that
+                    # snapshot as its source, resolved to a Snapshot object
+                    # (not a raw id/URI).
+                    self.assertIsNotNone(
+                        snap_vol.source,
+                        "A volume created from a snapshot must report a "
+                        "source")
+                    self.assertEqual(
+                        snap_vol.source.id, test_snap.id,
+                        "A volume's source should be the snapshot it was "
+                        "created from")
 
                 # Test volume creation from a snapshot (via Snapshot)
                 snap_vol2 = test_snap.create_volume()