8754846320

Committed 19 Apr 2024 02:02PM UTC coverage: 56.457% (-0.03%) from 56.485%

Build Type

push

github

Committed by Osvaldo Demo

Commit Message

properly parse source key in modules

Run Details

226 of 616 branches covered (36.69%)

Branch coverage included in aggregate %.

1 of 9 new or added lines in 1 file covered. (11.11%)

13 existing lines in 1 file now uncovered.

2598 of 4386 relevant lines covered (59.23%)

0.59 hits per line

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

26.97

/leverage/checker/version_parser.py

import json
import os
import re
from typing import Dict, Any, Union, List

import hcl2


class VersionExtractor:
    """Extracts versions from parsed Terraform configurations using best practices with type tagging."""

    @staticmethod
    def extract_versions(tf_config: Dict[str, Any]) -> Dict[str, Dict[str, str]]:
        versions = {}
        VersionExtractor.extract_core_and_providers(tf_config, versions)
        VersionExtractor.extract_module_versions(tf_config, versions)
        return versions

    @staticmethod
    def extract_core_and_providers(tf_config: Dict[str, Any], versions: Dict[str, Dict[str, str]]):
        for terraform_block in tf_config.get("terraform", []):
            if isinstance(terraform_block, dict):
                if "required_version" in terraform_block:
                    versions["terraform"] = {"type": "core", "version": terraform_block["required_version"]}
                if "required_providers" in terraform_block:
                    VersionExtractor.process_providers(terraform_block["required_providers"], versions)

    @staticmethod
    def process_providers(providers: Union[Dict[str, Any], List[Dict[str, Any]]], versions: Dict[str, Dict[str, str]]):
        if isinstance(providers, dict):
            VersionExtractor.extract_provider_versions(providers, versions)
        elif isinstance(providers, list):
            for provider_dict in providers:
                VersionExtractor.extract_provider_versions(provider_dict, versions)
        else:
            print(f"Error: Providers data structure not recognized: {providers}")

    @staticmethod
    def extract_provider_versions(providers: Dict[str, Any], versions: Dict[str, Dict[str, str]]):
        for provider, details in providers.items():
            if isinstance(details, dict) and "version" in details:
                versions[provider] = {"type": "provider", "version": details["version"]}
            elif isinstance(details, str):
                versions[provider] = {"type": "provider", "version": details}

    @staticmethod
    def extract_module_versions(tf_config: Dict[str, Any], versions: Dict[str, Dict[str, str]]):
        module_version_pattern = re.compile(r"\?ref=v([\d\.]+)$")
        for module in tf_config.get("module", []):
            if isinstance(module, dict) and len(module) == 1:
                for name, data in module.items():
                    source = data["source"]
                    match = module_version_pattern.search(source)
                    if match:
                        versions[name] = {"type": "module", "version": match.group(1), "repo": source.split("?")[0]}
                    elif "version" in module:
                        versions[name] = {"type": "module", "version": data["version"]}


class TerraformFileParser:
    @staticmethod
    def load(file_path: str) -> Dict[str, Any]:
        with open(file_path, "r") as tf_file:
            return hcl2.load(tf_file)


class VersionManager:
    def __init__(self, directory: str):
        self.directory = directory

    def find_versions(self) -> Dict[str, Dict[str, str]]:
        versions = {}
        for root, _, files in os.walk(self.directory):
            for file in files:
                if file.endswith(".tf"):
                    file_path = os.path.join(root, file)
                    try:
                        tf_config = TerraformFileParser.load(file_path)
                        versions.update(VersionExtractor.extract_versions(tf_config))
                    except Exception as e:
                        self.handle_error(e, file_path)
        return versions

    @staticmethod
    def handle_error(e: Exception, file_path: str):
        print(f"Error processing {file_path}: {e}")

1	import json	1✔
2	import os	1✔
3	import re	1✔
4	from typing import Dict, Any, Union, List	1✔
5
6	import hcl2	1✔
7
8
9	class VersionExtractor:	1✔
10	"""Extracts versions from parsed Terraform configurations using best practices with type tagging."""
11
12	@staticmethod	1✔
13	def extract_versions(tf_config: Dict[str, Any]) -> Dict[str, Dict[str, str]]:	1✔
14	versions = {}	×
15	VersionExtractor.extract_core_and_providers(tf_config, versions)	×
16	VersionExtractor.extract_module_versions(tf_config, versions)	×
17	return versions	×
18
19	@staticmethod	1✔
20	def extract_core_and_providers(tf_config: Dict[str, Any], versions: Dict[str, Dict[str, str]]):	1✔
21	for terraform_block in tf_config.get("terraform", []):	×
22	if isinstance(terraform_block, dict):	×
23	if "required_version" in terraform_block:	×
24	versions["terraform"] = {"type": "core", "version": terraform_block["required_version"]}	×
25	if "required_providers" in terraform_block:	×
26	VersionExtractor.process_providers(terraform_block["required_providers"], versions)	×
27
28	@staticmethod	1✔
29	def process_providers(providers: Union[Dict[str, Any], List[Dict[str, Any]]], versions: Dict[str, Dict[str, str]]):	1✔
30	if isinstance(providers, dict):	×
31	VersionExtractor.extract_provider_versions(providers, versions)	×
32	elif isinstance(providers, list):	×
33	for provider_dict in providers:	×
34	VersionExtractor.extract_provider_versions(provider_dict, versions)	×
35	else:
36	print(f"Error: Providers data structure not recognized: {providers}")	×
37
38	@staticmethod	1✔
39	def extract_provider_versions(providers: Dict[str, Any], versions: Dict[str, Dict[str, str]]):	1✔
40	for provider, details in providers.items():	×
41	if isinstance(details, dict) and "version" in details:	×
42	versions[provider] = {"type": "provider", "version": details["version"]}	×
43	elif isinstance(details, str):	×
44	versions[provider] = {"type": "provider", "version": details}	×
45
46	@staticmethod	1✔
47	def extract_module_versions(tf_config: Dict[str, Any], versions: Dict[str, Dict[str, str]]):	1✔
48	module_version_pattern = re.compile(r"\?ref=v([\d\.]+)$")	×
49	for module in tf_config.get("module", []):	×
NEW 50	if isinstance(module, dict) and len(module) == 1:	×
NEW 51	for name, data in module.items():	×
NEW 52	source = data["source"]	×
NEW 53	match = module_version_pattern.search(source)	×
NEW 54	if match:	×
NEW 55	versions[name] = {"type": "module", "version": match.group(1), "repo": source.split("?")[0]}	×
NEW 56	elif "version" in module:	×
NEW 57	versions[name] = {"type": "module", "version": data["version"]}	×
58
59
60	class TerraformFileParser:	1✔
61	@staticmethod	1✔
62	def load(file_path: str) -> Dict[str, Any]:	1✔
63	with open(file_path, "r") as tf_file:	×
64	return hcl2.load(tf_file)	×
65
66
67	class VersionManager:	1✔
68	def __init__(self, directory: str):	1✔
69	self.directory = directory	×
70
71	def find_versions(self) -> Dict[str, Dict[str, str]]:	1✔
72	versions = {}	×
73	for root, _, files in os.walk(self.directory):	×
74	for file in files:	×
75	if file.endswith(".tf"):	×
76	file_path = os.path.join(root, file)	×
77	try:	×
78	tf_config = TerraformFileParser.load(file_path)	×
79	versions.update(VersionExtractor.extract_versions(tf_config))	×
80	except Exception as e:	×
81	self.handle_error(e, file_path)	×
82	return versions	×
83
84	@staticmethod	1✔
85	def handle_error(e: Exception, file_path: str):	1✔
86	print(f"Error processing {file_path}: {e}")	×

binbashar / leverage / 8754846320

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