Add firewall to Pulumi #1375
Add firewall to Pulumi #1375
Conversation
jemrobinson
force-pushed
the
1374-add-firewall
branch
2 times, most recently
from
bdaedaa to
8841b9b
Compare
jemrobinson
force-pushed
the
1374-add-firewall
branch
from
5242115 to
62bf1b3
Compare
jemrobinson
force-pushed
the
1374-add-firewall
branch
from
62bf1b3 to
0508d34
Compare
… links to automation account.
…g removed on redeploy
jemrobinson
force-pushed
the
1374-add-firewall
branch
3 times, most recently
from
4f94b09 to
de9b59e
Compare
|
I think I've addressed all of @JimMadge's comments. I've checked that creation/teardown of SHM and SRE work as expected. This involved adding a few additional cross-referenced dependencies to ensure that resources are created in the correct order. There are also some general type-hinting fixes that I added while debugging. |
…ure) and reboot DC during SHM deployment
jemrobinson
force-pushed
the
1374-add-firewall
branch
from
de9b59e to
32ee60a
Compare
There was a problem hiding this comment.
This already got merged 😮.
Routing the Pulumi Inputs/Outputs and dealing with promises adds code, but I'm pretty confident it is a better solution overall. Should be more robust and it offloads handling state to Pulumi.
I don't like all the places that typing.cast has been introduced. It feels like sidestepping validating/sanitising input. Which will be necessary because we can't guarantee what a user or API will give to the program.
| @@ -115,7 +121,7 @@ def handle(self) -> None: | |||
| def add_missing_values(self, config: Config) -> None: | |||
| """Request any missing config values and add them to the config""" | |||
| # Request FQDN if not provided | |||
| fqdn = self.option("fqdn") | |||
| fqdn = cast(str, self.option("fqdn")) | |||
There was a problem hiding this comment.
Does this just trick the type checker that the result of self.option("fqdn") is always a string?
What if it is empty?
What if it isn't a string?
There was a problem hiding this comment.
This is a limitation in the type signature of Command.option. It returns dict[Unknown, Unknown] | Unknown | Any | Literal[False] which are all of the possibilities that cleo can coerce the command-line arguments into.
However, this isn't very useful for type checking. I'll have another think about this.
| @@ -535,7 +540,7 @@ def delete_application( | |||
| f"Could not delete application '{application_name}'.\n{str(exc)}" | |||
| ) from exc | |||
|
|
|||
| def get_application_by_name(self, application_name: str) -> JSONType: | |||
| def get_application_by_name(self, application_name: str) -> Dict[str, Any] | None: | |||
There was a problem hiding this comment.
| def get_application_by_name(self, application_name: str) -> Dict[str, Any] | None: | |
| def get_application_by_name(self, application_name: str) -> Optional[Dict[str, Any]]: |
| @@ -555,13 +562,16 @@ def get_service_principal_by_name(self, service_principal_name: str) -> JSONType | |||
| except (DataSafeHavenMicrosoftGraphException, IndexError): | |||
| return None | |||
|
|
|||
| def get_id_from_application_name(self, application_name: str) -> str: | |||
| def get_id_from_application_name(self, application_name: str) -> str | None: | |||
There was a problem hiding this comment.
| def get_id_from_application_name(self, application_name: str) -> str | None: | |
| def get_id_from_application_name(self, application_name: str) -> Optional[str]: |
| except DataSafeHavenMicrosoftGraphException: | ||
| return None | ||
|
|
||
| def get_id_from_groupname(self, group_name: str) -> str: | ||
| def get_id_from_groupname(self, group_name: str) -> str | None: |
There was a problem hiding this comment.
| def get_id_from_groupname(self, group_name: str) -> str | None: | |
| def get_id_from_groupname(self, group_name: str) -> Optional[str]: |
| @@ -571,7 +581,7 @@ def get_id_from_groupname(self, group_name: str) -> str: | |||
| except (DataSafeHavenMicrosoftGraphException, IndexError): | |||
| return None | |||
|
|
|||
| def get_id_from_username(self, username: str) -> str: | |||
| def get_id_from_username(self, username: str) -> str | None: | |||
There was a problem hiding this comment.
| def get_id_from_username(self, username: str) -> str | None: | |
| def get_id_from_username(self, username: str) -> Optional[str]: |
| @@ -664,7 +686,7 @@ def http_post(self, url: str, **kwargs: Any) -> requests.Response: | |||
| f"Could not execute POST request.\n{str(exc)}" | |||
| ) from exc | |||
|
|
|||
| def read_applications(self) -> JSONType: | |||
| def read_applications(self) -> Sequence[Dict[str, Any]]: | |||
There was a problem hiding this comment.
Do we really not know what concrete collection is returned here?
| def next_subnet(self, number_of_addresses: int) -> "AzureIPv4Range": | ||
| """Find the next unused subnet of a given size""" | ||
| if not math.log2(number_of_addresses).is_integer(): | ||
| raise DataSafeHavenIPRangeException( | ||
| f"Number of address '{number_of_addresses}' must be a power of 2" | ||
| ) | ||
| ip_address_first = self[0] | ||
| while True: | ||
| ip_address_last = ip_address_first + int(number_of_addresses - 1) | ||
| with suppress(DataSafeHavenIPRangeException): | ||
| candidate = AzureIPv4Range(ip_address_first, ip_address_last) | ||
| if not any(subnet.overlaps(candidate) for subnet in self.subnets): | ||
| self.subnets.append(candidate) | ||
| break | ||
| ip_address_first = ip_address_first + number_of_addresses | ||
| return candidate |
There was a problem hiding this comment.
This feels like an Advent of Code problem 😄.
There was a problem hiding this comment.
Yeah, sorry about that! If you think of a better alternative feel free to suggest one :)
| self.vm_size = vm_size | ||
|
|
||
| @property | ||
| def os_profile(self) -> compute.OSProfileArgs: | ||
| def os_profile(self) -> compute.OSProfileArgs | None: |
There was a problem hiding this comment.
| def os_profile(self) -> compute.OSProfileArgs | None: | |
| def os_profile(self) -> Optional[compute.OSProfileArgs]: |
| return self.os_profile_args | ||
|
|
||
| @property | ||
| def image_reference(self) -> compute.ImageReferenceArgs: | ||
| def image_reference(self) -> compute.ImageReferenceArgs | None: |
There was a problem hiding this comment.
| def image_reference(self) -> compute.ImageReferenceArgs | None: | |
| def image_reference(self) -> Optional[compute.ImageReferenceArgs]: |
✅ Checklist
Enable foobar integrationrather than515 foobar).developpython-migration branch../tests/AutoFormat_Powershell.ps1 -TargetPath <path to file or directory>.🌂 Related issues
Closes #1374 . Closes #1376. Closes #1384.
🔬 Tests
Tested on a local deployment