16530594834

Committed 25 Jul 2025 08:03PM UTC coverage: 93.735%. First build

Build # 16530594834

Build Type

Pull #142

github

Committed by

web-flow

Commit Message

Merge 084b83a28 into 863ada33f

Pull Request Pull Request #142: Nomalized Name Collisons

Run Details

36 of 38 new or added lines in 4 files covered. (94.74%)

2798 of 2985 relevant lines covered (93.74%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.77

/api/src/app/core/cdktf/ranges/base_range.py

import asyncio
import json
import logging
import os
import shutil
import uuid
from abc import ABC, abstractmethod
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import aiofiles
import aiofiles.os as aio_os
from cdktf import App

from ....enums.range_states import RangeState
from ....enums.regions import OpenLabsRegion
from ....schemas.range_schemas import (
    BlueprintRangeSchema,
    DeployedRangeCreateSchema,
    DeployedRangeSchema,
)
from ....schemas.secret_schema import SecretSchema
from ....utils.cdktf_utils import gen_resource_logical_ids
from ....utils.name_utils import normalize_name
from ...config import settings
from ..stacks.base_stack import AbstractBaseStack

# Configure logging
logger = logging.getLogger(__name__)


class AbstractBaseRange(ABC):
    """Abstract class to enforce common functionality across range cloud providers."""

    # Mutex for terraform init calls
    _init_lock = asyncio.Lock()

    # State variables
    _is_synthesized: bool
    _is_deployed: bool

    def __init__(  # noqa: PLR0913
        self,
        name: str,
        range_obj: BlueprintRangeSchema | DeployedRangeSchema,
        region: OpenLabsRegion,
        secrets: SecretSchema,
        description: str,
        state_file: dict[str, Any] | None = None,
    ) -> None:
        """Initialize CDKTF base range object."""
        self.name = name
        self.range_obj = range_obj
        self.region = region
        self.secrets = secrets
        self.description = description
        self.state_file = state_file
        if not self.state_file:
            self._is_deployed = False
        else:
            self._is_deployed = True

        # Initial values
        self.unique_str = uuid.uuid4()

        # Remove spaces to avoid CDKTF errors
        self.stack_name = f"{normalize_name(self.range_obj.name)}-{self.unique_str}"
        self._is_synthesized = False
        self.deployed_range_name = f"{normalize_name(self.name)}-{self.unique_str}"

    @abstractmethod
    def get_provider_stack_class(self) -> type[AbstractBaseStack]:
        """Return specific provider stack class to instantiate.

        Returns
        -------
        type[AbstractBaseStack]: Provider stack class.

        """
        pass

    @abstractmethod
    def has_secrets(self) -> bool:
        """Return if range has correct provider cloud credentials.

        Returns
        -------
            bool: True if correct provider creds exist. False otherwise.

        """
        pass

    @abstractmethod
    def get_cred_env_vars(self) -> dict[str, Any]:
        """Return dictionary of BASH Terraform cloud credential environment variables.

        Returns
        -------
            dict[str, Any]: Dict of BASH environment variables.

        """
        pass

    async def synthesize(self) -> bool:
        """Abstract method to synthesize terraform configuration.

        Returns:
            bool: True if successful synthesis. False otherwise.

        """
        try:
            logger.info("Synthesizing selected range: %s", self.name)

            # Create CDKTF app
            app = App(outdir=settings.CDKTF_DIR)

            # Instantiate the correct provider stack
            stack_class = self.get_provider_stack_class()
            stack_class(
                scope=app,
                range_obj=self.range_obj,
                cdktf_id=self.stack_name,
                cdktf_dir=settings.CDKTF_DIR,
                region=self.region,
                range_name=self.deployed_range_name,
            )

            # Synthesize Terraform files
            await asyncio.to_thread(app.synth)
            logger.info(
                "Range: %s synthesized successfully as stack: %s",
                self.name,
                self.stack_name,
            )

            self._is_synthesized = True
            return True
        except Exception as e:
            logger.error(
                "Error during synthesis of range: %s with stack: %s. Error: %s",
                self.name,
                self.stack_name,
                e,
            )
            return False

    async def _async_run_command(
        self, command: list[str], with_creds: bool = False
    ) -> tuple[str, str, int | None]:
        """Run a command asynchronously in the synth directory.

        Args:
        ----
            command (list[str]): Command to run in a list format.
            with_creds (bool): Include cloud credentials in execution environment.

        Returns:
        -------
            str: Standard output.
            str: Standard error.
            int | None: Return code of command if available.

        """
        synth_dir = self.get_synth_dir()
        if not await aio_os.path.exists(synth_dir):
            msg = f"Synthesis directory does not exist: {synth_dir}"
            raise FileNotFoundError(msg)

        env = os.environ.copy()
        if with_creds:
            env.update(self.get_cred_env_vars())

        process = await asyncio.create_subprocess_exec(
            *command,
            cwd=synth_dir,
            env=env,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout_bytes, stderr_bytes = await process.communicate()
        stdout, stderr = stdout_bytes.decode().strip(), stderr_bytes.decode().strip()

        if stdout:
            logger.info(
                "Command `cd %s && %s` stdout:\n%s",
                synth_dir,
                " ".join(command),
                stdout,
            )
        if stderr:
            logger.warning(
                "Command `cd %s && %s` stderr:\n%s",
                synth_dir,
                " ".join(command),
                stderr,
            )

        return stdout, stderr, process.returncode

    async def _init(self) -> bool:
        """Run `terraform init` programatically.

        Returns
        -------
            bool: True if successfully initialized. False otherwise.

        """
        async with AbstractBaseRange._init_lock:
            logger.info("Acquired lock for terraform init.")
            init_command = ["terraform", "init"]
            _, _, return_code = await self._async_run_command(
                init_command, with_creds=False
            )

            if return_code != 0:
                logger.error("Terraform init failed.")
                return False

            logger.info("Terraform init completed successfully.")

            return True

    async def deploy(self) -> DeployedRangeCreateSchema | None:
        """Run `terraform deploy --auto-approve` programmatically.

        Returns
        -------
            DeployedRangeCreateSchema: The deployed range creation schema to add to the database.

        """
        if not self.is_synthesized():
            logger.error("Range to destroy is not synthesized!")
            return None

        try:
            # Terraform init
            init_success = await self._init()
            if not init_success:
                msg = "Terraform init failed."
                raise RuntimeError(msg)

            # Terraform apply
            logger.info("Deploying selected range: %s", self.name)

            # Resources are deployed continously during apply command
            self._is_deployed = True

            apply_command = ["terraform", "apply", "--auto-approve"]
            _, _, return_code = await self._async_run_command(
                apply_command, with_creds=True
            )

            if return_code != 0:
                msg = "Terraform apply failed."
                raise RuntimeError(msg)

            # Load state
            if await aio_os.path.exists(self.get_state_file_path()):
                async with aiofiles.open(
                    self.get_state_file_path(), "r", encoding="utf-8"
                ) as f:
                    content = await f.read()
                    self.state_file = json.loads(content)
            else:
                msg = f"State file was not created during deployment. Expected path: {self.get_state_file_path()}"
                raise FileNotFoundError(msg)

            # Parse output variables
            deployed_range = await self._parse_terraform_outputs()
            if not deployed_range:
                msg = "Failed to parse terraform outputs!"
                raise RuntimeError(msg)
        except Exception as e:
            logger.exception("Error during deployment: %s", e)
            if self.is_deployed():
                destroy_success = await self.destroy()
                if not destroy_success:
                    logger.critical(
                        "Failed to cleanup after deployment failure of range: %s",
                        self.name,
                    )
            return None
        finally:
            # Delete files made during deployment
            await self.cleanup_synth()

        logger.info("Successfully deployed range: %s", self.name)

        return deployed_range

    async def destroy(self) -> bool:
        """Destroy terraform infrastructure.

        Returns
        -------
            bool: True if destroy was successful. False otherwise.

        """
        if not self.is_deployed():
            logger.error("Can't destroy range that is not deployed!")
            return False

        if not self.is_synthesized():
            logger.error("Range to destory is not synthesized!")
            return False

        try:
            # Try to create the state file
            created_state_file = await self.create_state_file()
            if not created_state_file:
                logger.info(
                    "State file not saved! Unable to create a new one in the filesystem."
                )

            # Check for an existing state file to be used for destroying
            if not await aio_os.path.exists(self.get_state_file_path()):
                msg = f"Unable to destroy range: {self.name} missing state file!"
                raise FileNotFoundError(msg)

            # Terraform init
            init_success = await self._init()
            if not init_success:
                msg = "Terraform init failed."
                raise RuntimeError(msg)

            # Terraform destroy
            logger.info(
                "Tearing down selected range: %s",
                self.name,
            )
            destroy_command = ["terraform", "destroy", "--auto-approve"]
            _, _, return_code = await self._async_run_command(
                destroy_command, with_creds=True
            )

            if return_code != 0:
                msg = "Terraform destroy failed"
                raise RuntimeError(msg)

            self._is_deployed = False

            logger.info("Successfully destroyed range: %s", self.name)
            return True
        except Exception as e:
            logger.exception("Error during destroy: %s", e)
            return False
        finally:
            # Delete synth files
            await self.cleanup_synth()

    async def _parse_terraform_outputs(  # noqa: PLR0911
        self,
    ) -> DeployedRangeCreateSchema | None:
        """Parse Terraform output variables into a deployed range object.

        Internal class function should not be called externally.

        Returns
        -------
            DeployedRangeCreateSchema: The terraform deploy output rendered as a desployed range pydantic schema.

        """
        if not await aio_os.path.exists(self.get_state_file_path()):
            logger.error(
                "Failed to find state file at: %s when attempting to parse terraform outputs.",
                self.get_state_file_path(),
            )
            return None

        try:
            output_command = ["terraform", "output", "-json"]
            stdout, _, return_code = await self._async_run_command(
                output_command, with_creds=False
            )

            if return_code != 0 or not stdout:
                msg = "Terraform output failed"
                raise RuntimeError(msg)

            logger.info("Terraform output completed successfully.")

        except Exception as e:
            logger.exception("Failed to parse Terraform outputs: %s", e)
            return None

        # Parse Terraform Output variables
        raw_outputs = json.loads(stdout)
        dumped_schema = self.range_obj.model_dump()

        try:
            # Range attributes
            jumpbox_key = next(
                (key for key in raw_outputs if key.endswith("-JumpboxInstanceId")),
                None,
            )
            jumpbox_ip_key = next(
                (key for key in raw_outputs if key.endswith("-JumpboxPublicIp")),
                None,
            )
            private_key = next(
                (key for key in raw_outputs if key.endswith("-private-key")),
                None,
            )

            if not all([jumpbox_key, jumpbox_ip_key, private_key]):
                logger.error(
                    "Could not find required keys in Terraform output: %s",
                    raw_outputs.keys(),
                )
                return None

            dumped_schema["jumpbox_resource_id"] = raw_outputs[jumpbox_key]["value"]
            dumped_schema["jumpbox_public_ip"] = raw_outputs[jumpbox_ip_key]["value"]
            dumped_schema["range_private_key"] = raw_outputs[private_key]["value"]

            vpc_logical_ids = gen_resource_logical_ids(
                [vpc.name for vpc in self.range_obj.vpcs]
            )
            for x, vpc in enumerate(self.range_obj.vpcs):
                vpc_logical_id = vpc_logical_ids[vpc.name]
                current_vpc = dumped_schema["vpcs"][x]

                vpc_key = next(
                    (
                        key
                        for key in raw_outputs
                        if key.endswith(f"-{vpc_logical_id}-resource-id")
                    ),
                    None,
                )
                if not vpc_key:
                    logger.error(
                        "Could not find VPC resource ID key for %s in Terraform output",
                        vpc_logical_id,
                    )
                    return None
                current_vpc["resource_id"] = raw_outputs[vpc_key]["value"]

                subnet_logical_ids = gen_resource_logical_ids(
                    [subnet.name for subnet in vpc.subnets]  # type: ignore
                )
                for y, subnet in enumerate(vpc.subnets):  # type: ignore
                    subnet_logical_id = subnet_logical_ids[subnet.name]
                    current_subnet = current_vpc["subnets"][y]

                    subnet_key = next(
                        (
                            key
                            for key in raw_outputs
                            if key.endswith(
                                f"-{vpc_logical_id}-{subnet_logical_id}-resource-id"
                            )
                        ),
                        None,
                    )
                    if not subnet_key:
                        logger.error(
                            "Could not find subnet resource ID key for %s in %s in Terraform output",
                            subnet_logical_id,
                            vpc_logical_id,
                        )
                        return None
                    current_subnet["resource_id"] = raw_outputs[subnet_key]["value"]

                    for z, host in enumerate(subnet.hosts):
                        current_host = current_subnet["hosts"][z]
                        host_id_key = next(
                            (
                                key
                                for key in raw_outputs
                                if key.endswith(
                                    f"-{vpc_logical_id}-{subnet_logical_id}-{host.hostname}-resource-id"
                                )
                            ),
                            None,
                        )
                        host_ip_key = next(
                            (
                                key
                                for key in raw_outputs
                                if key.endswith(
                                    f"-{vpc_logical_id}-{subnet_logical_id}-{host.hostname}-private-ip"
                                )
                            ),
                            None,
                        )

                        if not host_id_key or not host_ip_key:
                            logger.error(
                                "Could not find host keys for %s in %s/%s in Terraform output",
                                host.hostname,
                                vpc_logical_id,
                                subnet_logical_id,
                            )
                            return None

                        current_host["resource_id"] = raw_outputs[host_id_key]["value"]
                        current_host["ip_address"] = raw_outputs[host_ip_key]["value"]
        except KeyError as e:
            logger.exception(
                "Failed to parse Terraform outputs. Missing key in output. Exception: %s",
                e,
            )
            return None
        except Exception as e:
            logger.exception(
                "Unknown error parsing Terraform outputs. Exception: %s", e
            )
            return None

        # Add missing attributes
        dumped_schema["name"] = self.name
        dumped_schema["description"] = self.description
        dumped_schema["date"] = datetime.now(tz=timezone.utc)
        dumped_schema["readme"] = None
        dumped_schema["state_file"] = self.get_state_file()
        dumped_schema["state"] = RangeState.ON
        dumped_schema["region"] = self.region

        return DeployedRangeCreateSchema.model_validate(dumped_schema)

    def is_synthesized(self) -> bool:
        """Return if range is currently synthesized."""
        return self._is_synthesized

    def is_deployed(self) -> bool:
        """Return if range is currently deployed."""
        return self._is_deployed

    def get_synth_dir(self) -> Path:
        """Get CDKTF synthesis directory."""
        return Path(f"{settings.CDKTF_DIR}/stacks/{self.stack_name}")

    def get_synth_file_path(self) -> Path:
        """Get path to terraform HCL from CDKTF synthesis."""
        return Path(f"{self.get_synth_dir()}/cdk.tf.json")

    def get_state_file(self) -> dict[str, Any] | None:
        """Return state file content.

        Range must have been deployed or have a state file passed on object creation.
        """
        return self.state_file

    def get_state_file_path(self) -> Path:
        """Get CDKTF state file path."""
        return self.get_synth_dir() / f"terraform.{self.stack_name}.tfstate"

    async def create_state_file(self) -> bool:
        """Create state file with contents.

        Range must have been deployed or have a state file passed on object creation.

        Returns
        -------
            bool: True if state file successfully written. False otherwise.

        """
        if not self.state_file:
            msg = f"Can't write state file none exists! Attempted on range: {self.name}"
            logger.warning(msg)
            return False

        json_string = json.dumps(self.state_file, indent=4)

        async with aiofiles.open(self.get_state_file_path(), mode="w") as file:
            await file.write(json_string)

        msg = f"Successfully created state file: {self.get_state_file_path()} "
        logger.info(msg)
        return True

    async def cleanup_synth(self) -> bool:
        """Delete Terraform files generated by CDKTF synthesis."""
        try:
            await asyncio.to_thread(shutil.rmtree, self.get_synth_dir())
            self._is_synthesized = False
            return True
        except Exception as e:
            logger.error(
                "Failed to delete synthesis files for stack: %s. Error: %s",
                self.stack_name,
                e,
            )
            return False

1	import asyncio	1✔
2	import json	1✔
3	import logging	1✔
4	import os	1✔
5	import shutil	1✔
6	import uuid	1✔
7	from abc import ABC, abstractmethod	1✔
8	from datetime import datetime, timezone	1✔
9	from pathlib import Path	1✔
10	from typing import Any	1✔
11
12	import aiofiles	1✔
13	import aiofiles.os as aio_os	1✔
14	from cdktf import App	1✔
15
16	from ....enums.range_states import RangeState	1✔
17	from ....enums.regions import OpenLabsRegion	1✔
18	from ....schemas.range_schemas import (	1✔
19	BlueprintRangeSchema,
20	DeployedRangeCreateSchema,
21	DeployedRangeSchema,
22	)
23	from ....schemas.secret_schema import SecretSchema	1✔
24	from ....utils.cdktf_utils import gen_resource_logical_ids	1✔
25	from ....utils.name_utils import normalize_name	1✔
26	from ...config import settings	1✔
27	from ..stacks.base_stack import AbstractBaseStack	1✔
28
29	# Configure logging
30	logger = logging.getLogger(__name__)	1✔
31
32
33	class AbstractBaseRange(ABC):	1✔
34	"""Abstract class to enforce common functionality across range cloud providers."""
35
36	# Mutex for terraform init calls
37	_init_lock = asyncio.Lock()	1✔
38
39	# State variables
40	_is_synthesized: bool	1✔
41	_is_deployed: bool	1✔
42
43	def __init__( # noqa: PLR0913	1✔
44	self,
45	name: str,
46	range_obj: BlueprintRangeSchema \| DeployedRangeSchema,
47	region: OpenLabsRegion,
48	secrets: SecretSchema,
49	description: str,
50	state_file: dict[str, Any] \| None = None,
51	) -> None:
52	"""Initialize CDKTF base range object."""
53	self.name = name	1✔
54	self.range_obj = range_obj	1✔
55	self.region = region	1✔
56	self.secrets = secrets	1✔
57	self.description = description	1✔
58	self.state_file = state_file	1✔
59	if not self.state_file:	1✔
60	self._is_deployed = False	1✔
61	else:
62	self._is_deployed = True	1✔
63
64	# Initial values
65	self.unique_str = uuid.uuid4()	1✔
66
67	# Remove spaces to avoid CDKTF errors
68	self.stack_name = f"{normalize_name(self.range_obj.name)}-{self.unique_str}"	1✔
69	self._is_synthesized = False	1✔
70	self.deployed_range_name = f"{normalize_name(self.name)}-{self.unique_str}"	1✔
71
72	@abstractmethod	1✔
73	def get_provider_stack_class(self) -> type[AbstractBaseStack]:	1✔
74	"""Return specific provider stack class to instantiate.
75
76	Returns
77	-------
78	type[AbstractBaseStack]: Provider stack class.
79
80	"""
81	pass	×
82
83	@abstractmethod	1✔
84	def has_secrets(self) -> bool:	1✔
85	"""Return if range has correct provider cloud credentials.
86
87	Returns
88	-------
89	bool: True if correct provider creds exist. False otherwise.
90
91	"""
92	pass	×
93
94	@abstractmethod	1✔
95	def get_cred_env_vars(self) -> dict[str, Any]:	1✔
96	"""Return dictionary of BASH Terraform cloud credential environment variables.
97
98	Returns
99	-------
100	dict[str, Any]: Dict of BASH environment variables.
101
102	"""
103	pass	×
104
105	async def synthesize(self) -> bool:	1✔
106	"""Abstract method to synthesize terraform configuration.
107
108	Returns:
109	bool: True if successful synthesis. False otherwise.
110
111	"""
112	try:	1✔
113	logger.info("Synthesizing selected range: %s", self.name)	1✔
114
115	# Create CDKTF app
116	app = App(outdir=settings.CDKTF_DIR)	1✔
117
118	# Instantiate the correct provider stack
119	stack_class = self.get_provider_stack_class()	1✔
120	stack_class(	1✔
121	scope=app,
122	range_obj=self.range_obj,
123	cdktf_id=self.stack_name,
124	cdktf_dir=settings.CDKTF_DIR,
125	region=self.region,
126	range_name=self.deployed_range_name,
127	)
128
129	# Synthesize Terraform files
130	await asyncio.to_thread(app.synth)	1✔
131	logger.info(	1✔
132	"Range: %s synthesized successfully as stack: %s",
133	self.name,
134	self.stack_name,
135	)
136
137	self._is_synthesized = True	1✔
138	return True	1✔
139	except Exception as e:	1✔
140	logger.error(	1✔
141	"Error during synthesis of range: %s with stack: %s. Error: %s",
142	self.name,
143	self.stack_name,
144	e,
145	)
146	return False	1✔
147
148	async def _async_run_command(	1✔
149	self, command: list[str], with_creds: bool = False
150	) -> tuple[str, str, int \| None]:
151	"""Run a command asynchronously in the synth directory.
152
153	Args:
154	----
155	command (list[str]): Command to run in a list format.
156	with_creds (bool): Include cloud credentials in execution environment.
157
158	Returns:
159	-------
160	str: Standard output.
161	str: Standard error.
162	int \| None: Return code of command if available.
163
164	"""
165	synth_dir = self.get_synth_dir()	1✔
166	if not await aio_os.path.exists(synth_dir):	1✔
167	msg = f"Synthesis directory does not exist: {synth_dir}"	1✔
168	raise FileNotFoundError(msg)	1✔
169
170	env = os.environ.copy()	1✔
171	if with_creds:	1✔
172	env.update(self.get_cred_env_vars())	1✔
173
174	process = await asyncio.create_subprocess_exec(	1✔
175	*command,
176	cwd=synth_dir,
177	env=env,
178	stdout=asyncio.subprocess.PIPE,
179	stderr=asyncio.subprocess.PIPE,
180	)
181	stdout_bytes, stderr_bytes = await process.communicate()	1✔
182	stdout, stderr = stdout_bytes.decode().strip(), stderr_bytes.decode().strip()	1✔
183
184	if stdout:	1✔
185	logger.info(	1✔
186	"Command `cd %s && %s` stdout:\n%s",
187	synth_dir,
188	" ".join(command),
189	stdout,
190	)
191	if stderr:	1✔
192	logger.warning(	1✔
193	"Command `cd %s && %s` stderr:\n%s",
194	synth_dir,
195	" ".join(command),
196	stderr,
197	)
198
199	return stdout, stderr, process.returncode	1✔
200
201	async def _init(self) -> bool:	1✔
202	"""Run `terraform init` programatically.
203
204	Returns
205	-------
206	bool: True if successfully initialized. False otherwise.
207
208	"""
209	async with AbstractBaseRange._init_lock:	1✔
210	logger.info("Acquired lock for terraform init.")	1✔
211	init_command = ["terraform", "init"]	1✔
212	_, _, return_code = await self._async_run_command(	1✔
213	init_command, with_creds=False
214	)
215
216	if return_code != 0:	1✔
217	logger.error("Terraform init failed.")	1✔
218	return False	1✔
219
220	logger.info("Terraform init completed successfully.")	1✔
221
222	return True	1✔
223
224	async def deploy(self) -> DeployedRangeCreateSchema \| None:	1✔
225	"""Run `terraform deploy --auto-approve` programmatically.
226
227	Returns
228	-------
229	DeployedRangeCreateSchema: The deployed range creation schema to add to the database.
230
231	"""
232	if not self.is_synthesized():	1✔
233	logger.error("Range to destroy is not synthesized!")	1✔
234	return None	1✔
235
236	try:	1✔
237	# Terraform init
238	init_success = await self._init()	1✔
239	if not init_success:	1✔
240	msg = "Terraform init failed."	1✔
241	raise RuntimeError(msg)	1✔
242
243	# Terraform apply
244	logger.info("Deploying selected range: %s", self.name)	1✔
245
246	# Resources are deployed continously during apply command
247	self._is_deployed = True	1✔
248
249	apply_command = ["terraform", "apply", "--auto-approve"]	1✔
250	_, _, return_code = await self._async_run_command(	1✔
251	apply_command, with_creds=True
252	)
253
254	if return_code != 0:	1✔
255	msg = "Terraform apply failed."	1✔
256	raise RuntimeError(msg)	1✔
257
258	# Load state
259	if await aio_os.path.exists(self.get_state_file_path()):	1✔
260	async with aiofiles.open(	1✔
261	self.get_state_file_path(), "r", encoding="utf-8"
262	) as f:
263	content = await f.read()	1✔
264	self.state_file = json.loads(content)	1✔
265	else:
266	msg = f"State file was not created during deployment. Expected path: {self.get_state_file_path()}"	1✔
267	raise FileNotFoundError(msg)	1✔
268
269	# Parse output variables
270	deployed_range = await self._parse_terraform_outputs()	1✔
271	if not deployed_range:	1✔
272	msg = "Failed to parse terraform outputs!"	1✔
273	raise RuntimeError(msg)	1✔
274	except Exception as e:	1✔
275	logger.exception("Error during deployment: %s", e)	1✔
276	if self.is_deployed():	1✔
277	destroy_success = await self.destroy()	1✔
278	if not destroy_success:	1✔
279	logger.critical(	1✔
280	"Failed to cleanup after deployment failure of range: %s",
281	self.name,
282	)
283	return None	1✔
284	finally:
285	# Delete files made during deployment
286	await self.cleanup_synth()	1✔
287
288	logger.info("Successfully deployed range: %s", self.name)	1✔
289
290	return deployed_range	1✔
291
292	async def destroy(self) -> bool:	1✔
293	"""Destroy terraform infrastructure.
294
295	Returns
296	-------
297	bool: True if destroy was successful. False otherwise.
298
299	"""
300	if not self.is_deployed():	1✔
301	logger.error("Can't destroy range that is not deployed!")	1✔
302	return False	1✔
303
304	if not self.is_synthesized():	1✔
305	logger.error("Range to destory is not synthesized!")	1✔
306	return False	1✔
307
308	try:	1✔
309	# Try to create the state file
310	created_state_file = await self.create_state_file()	1✔
311	if not created_state_file:	1✔
312	logger.info(	1✔
313	"State file not saved! Unable to create a new one in the filesystem."
314	)
315
316	# Check for an existing state file to be used for destroying
317	if not await aio_os.path.exists(self.get_state_file_path()):	1✔
318	msg = f"Unable to destroy range: {self.name} missing state file!"	1✔
319	raise FileNotFoundError(msg)	1✔
320
321	# Terraform init
322	init_success = await self._init()	1✔
323	if not init_success:	1✔
324	msg = "Terraform init failed."	1✔
325	raise RuntimeError(msg)	1✔
326
327	# Terraform destroy
328	logger.info(	1✔
329	"Tearing down selected range: %s",
330	self.name,
331	)
332	destroy_command = ["terraform", "destroy", "--auto-approve"]	1✔
333	_, _, return_code = await self._async_run_command(	1✔
334	destroy_command, with_creds=True
335	)
336
337	if return_code != 0:	1✔
338	msg = "Terraform destroy failed"	1✔
339	raise RuntimeError(msg)	1✔
340
341	self._is_deployed = False	1✔
342
343	logger.info("Successfully destroyed range: %s", self.name)	1✔
344	return True	1✔
345	except Exception as e:	1✔
346	logger.exception("Error during destroy: %s", e)	1✔
347	return False	1✔
348	finally:
349	# Delete synth files
350	await self.cleanup_synth()	1✔
351
352	async def _parse_terraform_outputs( # noqa: PLR0911	1✔
353	self,
354	) -> DeployedRangeCreateSchema \| None:
355	"""Parse Terraform output variables into a deployed range object.
356
357	Internal class function should not be called externally.
358
359	Returns
360	-------
361	DeployedRangeCreateSchema: The terraform deploy output rendered as a desployed range pydantic schema.
362
363	"""
364	if not await aio_os.path.exists(self.get_state_file_path()):	1✔
365	logger.error(	1✔
366	"Failed to find state file at: %s when attempting to parse terraform outputs.",
367	self.get_state_file_path(),
368	)
369	return None	1✔
370
371	try:	1✔
372	output_command = ["terraform", "output", "-json"]	1✔
373	stdout, _, return_code = await self._async_run_command(	1✔
374	output_command, with_creds=False
375	)
376
377	if return_code != 0 or not stdout:	1✔
378	msg = "Terraform output failed"	1✔
379	raise RuntimeError(msg)	1✔
380
381	logger.info("Terraform output completed successfully.")	1✔
382
383	except Exception as e:	1✔
384	logger.exception("Failed to parse Terraform outputs: %s", e)	1✔
385	return None	1✔
386
387	# Parse Terraform Output variables
388	raw_outputs = json.loads(stdout)	1✔
389	dumped_schema = self.range_obj.model_dump()	1✔
390
391	try:	1✔
392	# Range attributes
393	jumpbox_key = next(	1✔
394	(key for key in raw_outputs if key.endswith("-JumpboxInstanceId")),
395	None,
396	)
397	jumpbox_ip_key = next(	1✔
398	(key for key in raw_outputs if key.endswith("-JumpboxPublicIp")),
399	None,
400	)
401	private_key = next(	1✔
402	(key for key in raw_outputs if key.endswith("-private-key")),
403	None,
404	)
405
406	if not all([jumpbox_key, jumpbox_ip_key, private_key]):	1✔
407	logger.error(	×
408	"Could not find required keys in Terraform output: %s",
409	raw_outputs.keys(),
410	)
411	return None	×
412
413	dumped_schema["jumpbox_resource_id"] = raw_outputs[jumpbox_key]["value"]	1✔
414	dumped_schema["jumpbox_public_ip"] = raw_outputs[jumpbox_ip_key]["value"]	1✔
415	dumped_schema["range_private_key"] = raw_outputs[private_key]["value"]	1✔
416
417	vpc_logical_ids = gen_resource_logical_ids(	1✔
418	[vpc.name for vpc in self.range_obj.vpcs]
419	)
420	for x, vpc in enumerate(self.range_obj.vpcs):	1✔
421	vpc_logical_id = vpc_logical_ids[vpc.name]	1✔
422	current_vpc = dumped_schema["vpcs"][x]	1✔
423
424	vpc_key = next(	×
425	(
426	key
427	for key in raw_outputs
428	if key.endswith(f"-{vpc_logical_id}-resource-id")
429	),
430	None,
431	)
432	if not vpc_key:	×
433	logger.error(	×
434	"Could not find VPC resource ID key for %s in Terraform output",
435	vpc_logical_id,
436	)
437	return None	×
438	current_vpc["resource_id"] = raw_outputs[vpc_key]["value"]	×
439
NEW 440	subnet_logical_ids = gen_resource_logical_ids(	×
441	[subnet.name for subnet in vpc.subnets] # type: ignore
442	)
443	for y, subnet in enumerate(vpc.subnets): # type: ignore	×
NEW 444	subnet_logical_id = subnet_logical_ids[subnet.name]	×
445	current_subnet = current_vpc["subnets"][y]	×
446
447	subnet_key = next(	×
448	(
449	key
450	for key in raw_outputs
451	if key.endswith(
452	f"-{vpc_logical_id}-{subnet_logical_id}-resource-id"
453	)
454	),
455	None,
456	)
457	if not subnet_key:	×
458	logger.error(	×
459	"Could not find subnet resource ID key for %s in %s in Terraform output",
460	subnet_logical_id,
461	vpc_logical_id,
462	)
463	return None	×
464	current_subnet["resource_id"] = raw_outputs[subnet_key]["value"]	×
465
466	for z, host in enumerate(subnet.hosts):	×
467	current_host = current_subnet["hosts"][z]	×
468	host_id_key = next(	×
469	(
470	key
471	for key in raw_outputs
472	if key.endswith(
473	f"-{vpc_logical_id}-{subnet_logical_id}-{host.hostname}-resource-id"
474	)
475	),
476	None,
477	)
478	host_ip_key = next(	×
479	(
480	key
481	for key in raw_outputs
482	if key.endswith(
483	f"-{vpc_logical_id}-{subnet_logical_id}-{host.hostname}-private-ip"
484	)
485	),
486	None,
487	)
488
489	if not host_id_key or not host_ip_key:	×
490	logger.error(	×
491	"Could not find host keys for %s in %s/%s in Terraform output",
492	host.hostname,
493	vpc_logical_id,
494	subnet_logical_id,
495	)
496	return None	×
497
498	current_host["resource_id"] = raw_outputs[host_id_key]["value"]	×
499	current_host["ip_address"] = raw_outputs[host_ip_key]["value"]	×
500	except KeyError as e:	1✔
501	logger.exception(	1✔
502	"Failed to parse Terraform outputs. Missing key in output. Exception: %s",
503	e,
504	)
505	return None	1✔
506	except Exception as e:	1✔
507	logger.exception(	1✔
508	"Unknown error parsing Terraform outputs. Exception: %s", e
509	)
510	return None	1✔
511
512	# Add missing attributes
513	dumped_schema["name"] = self.name	×
514	dumped_schema["description"] = self.description	×
515	dumped_schema["date"] = datetime.now(tz=timezone.utc)	×
516	dumped_schema["readme"] = None	×
517	dumped_schema["state_file"] = self.get_state_file()	×
518	dumped_schema["state"] = RangeState.ON	×
519	dumped_schema["region"] = self.region	×
520
521	return DeployedRangeCreateSchema.model_validate(dumped_schema)	×
522
523	def is_synthesized(self) -> bool:	1✔
524	"""Return if range is currently synthesized."""
525	return self._is_synthesized	1✔
526
527	def is_deployed(self) -> bool:	1✔
528	"""Return if range is currently deployed."""
529	return self._is_deployed	1✔
530
531	def get_synth_dir(self) -> Path:	1✔
532	"""Get CDKTF synthesis directory."""
533	return Path(f"{settings.CDKTF_DIR}/stacks/{self.stack_name}")	1✔
534
535	def get_synth_file_path(self) -> Path:	1✔
536	"""Get path to terraform HCL from CDKTF synthesis."""
537	return Path(f"{self.get_synth_dir()}/cdk.tf.json")	1✔
538
539	def get_state_file(self) -> dict[str, Any] \| None:	1✔
540	"""Return state file content.
541
542	Range must have been deployed or have a state file passed on object creation.
543	"""
544	return self.state_file	1✔
545
546	def get_state_file_path(self) -> Path:	1✔
547	"""Get CDKTF state file path."""
548	return self.get_synth_dir() / f"terraform.{self.stack_name}.tfstate"	1✔
549
550	async def create_state_file(self) -> bool:	1✔
551	"""Create state file with contents.
552
553	Range must have been deployed or have a state file passed on object creation.
554
555	Returns
556	-------
557	bool: True if state file successfully written. False otherwise.
558
559	"""
560	if not self.state_file:	1✔
561	msg = f"Can't write state file none exists! Attempted on range: {self.name}"	1✔
562	logger.warning(msg)	1✔
563	return False	1✔
564
565	json_string = json.dumps(self.state_file, indent=4)	1✔
566
567	async with aiofiles.open(self.get_state_file_path(), mode="w") as file:	1✔
568	await file.write(json_string)	1✔
569
570	msg = f"Successfully created state file: {self.get_state_file_path()} "	1✔
571	logger.info(msg)	1✔
572	return True	1✔
573
574	async def cleanup_synth(self) -> bool:	1✔
575	"""Delete Terraform files generated by CDKTF synthesis."""
576	try:	1✔
577	await asyncio.to_thread(shutil.rmtree, self.get_synth_dir())	1✔
578	self._is_synthesized = False	1✔
579	return True	1✔
580	except Exception as e:	1✔
581	logger.error(	1✔
582	"Failed to delete synthesis files for stack: %s. Error: %s",
583	self.stack_name,
584	e,
585	)
586	return False	1✔

OpenLabsHQ / OpenLabs / 16530594834

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous