18854987527

Committed 27 Oct 2025 08:28PM UTC coverage: 80.311% (+0.2%) from 80.141%

Build # 18854987527

Build Type

Pull #112

github

Committed by

web-flow

Commit Message

Merge efeaa0bed into 3ef0e3272

Pull Request Pull Request #112: fix: multipass entries

Run Details

213 of 293 branches covered (72.7%)

Branch coverage included in aggregate %.

42 of 55 new or added lines in 15 files covered. (76.36%)

2 existing lines in 2 files now uncovered.

2418 of 2983 relevant lines covered (81.06%)

0.81 hits per line

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

82.0

/src/socm/utils/misc.py

import ast
from typing import Dict, List


def parse_comma_separated_fields(config: dict, fields_to_parse: List[str]) -> dict:
    """Convert comma-separated string values to lists."""
    for key, value in config.items():
        if isinstance(value, dict):
            parse_comma_separated_fields(value, fields_to_parse)
        elif key in fields_to_parse and isinstance(value, str) and ',' in value:
            config[key] = [ast.literal_eval(item.strip()) for item in value.split(',')]
    return config

def get_workflow_entries(campaign_dict: dict, subcampaign_map: Dict[str, list] | None = None) -> Dict[str, dict]:
    """
    Extract workflow entries from a campaign dictionary using a predefined mapping.

    Args:
        campaign_dict: A dictionary containing campaign configuration
        subcampaign_map: A dictionary mapping subcampaign names to lists of their workflow names
                         E.g., {"ml-null-test": ["mission-tests", "wafer-tests"]}

    Returns:
        Dictionary containing workflow entries
    """
    campaign_data = campaign_dict.get("campaign", {})

    # Default empty map if none provided
    if subcampaign_map is None:
        subcampaign_map = {}

    # Collect all workflows (direct and from subcampaigns)
    workflows = {}

    for workflow_key, workflow_value in campaign_data.items():
        # Skip non-dictionary values
        if not isinstance(workflow_value, dict):
            continue

        # Check if this is a known subcampaign
        if workflow_key in subcampaign_map:
            # Process known workflows for this subcampaign
            subcampaign_name = workflow_key
            subcampaign_workflows = subcampaign_map[workflow_key]

            # Create a copy of the subcampaign config without its workflows
            subcampaign_common_config = {
                k: v for k, v in workflow_value.items() if k not in subcampaign_workflows
            }

            for workflow_name in subcampaign_workflows:
                if workflow_name in workflow_value:
                    # Start with the workflow's own config
                    workflow_config = workflow_value[workflow_name].copy()

                    # Update with common subcampaign config
                    workflow_config.update(subcampaign_common_config)

                    if isinstance(workflow_config, dict):
                        # Create combined key: subcampaign.workflow_name
                        workflows[f"{subcampaign_name}.{workflow_name}"] = (
                            workflow_config
                        )
        else:
            # Treat as regular workflow
            workflows[workflow_key] = workflow_value

    return workflows


def get_query_from_file(file_path: str) -> str:
    """
    Extract a SQL query from a file.

    Args:
        file_path: The path to the file containing the SQL query.

    Returns:
        The SQL query as a string.
    """
    query = "obs_id IN ("
    with open(file_path, "r") as file:
        obslist = file.readlines()
        for obs_id in obslist:
            obs_id = obs_id.strip()
            query += f"'{obs_id}',"
    query = query.rstrip(",")
    query += ")"

    return query

1	import ast	1✔
2	from typing import Dict, List	1✔
3
4
5	def parse_comma_separated_fields(config: dict, fields_to_parse: List[str]) -> dict:	1✔
6	"""Convert comma-separated string values to lists."""
NEW 7	for key, value in config.items():	×
NEW 8	if isinstance(value, dict):	×
NEW 9	parse_comma_separated_fields(value, fields_to_parse)	×
NEW 10	elif key in fields_to_parse and isinstance(value, str) and ',' in value:	×
NEW 11	config[key] = [ast.literal_eval(item.strip()) for item in value.split(',')]	×
NEW 12	return config	×
13
14	def get_workflow_entries(campaign_dict: dict, subcampaign_map: Dict[str, list] \| None = None) -> Dict[str, dict]:	1✔
15	"""
16	Extract workflow entries from a campaign dictionary using a predefined mapping.
17
18	Args:
19	campaign_dict: A dictionary containing campaign configuration
20	subcampaign_map: A dictionary mapping subcampaign names to lists of their workflow names
21	E.g., {"ml-null-test": ["mission-tests", "wafer-tests"]}
22
23	Returns:
24	Dictionary containing workflow entries
25	"""
26	campaign_data = campaign_dict.get("campaign", {})	1✔
27
28	# Default empty map if none provided
29	if subcampaign_map is None:	1✔
30	subcampaign_map = {}	1✔
31
32	# Collect all workflows (direct and from subcampaigns)
33	workflows = {}	1✔
34
35	for workflow_key, workflow_value in campaign_data.items():	1✔
36	# Skip non-dictionary values
37	if not isinstance(workflow_value, dict):	1✔
38	continue	1✔
39
40	# Check if this is a known subcampaign
41	if workflow_key in subcampaign_map:	1✔
42	# Process known workflows for this subcampaign
43	subcampaign_name = workflow_key	1✔
44	subcampaign_workflows = subcampaign_map[workflow_key]	1✔
45
46	# Create a copy of the subcampaign config without its workflows
47	subcampaign_common_config = {	1✔
48	k: v for k, v in workflow_value.items() if k not in subcampaign_workflows
49	}
50
51	for workflow_name in subcampaign_workflows:	1✔
52	if workflow_name in workflow_value:	1✔
53	# Start with the workflow's own config
54	workflow_config = workflow_value[workflow_name].copy()	1✔
55
56	# Update with common subcampaign config
57	workflow_config.update(subcampaign_common_config)	1✔
58
59	if isinstance(workflow_config, dict):	1✔
60	# Create combined key: subcampaign.workflow_name
61	workflows[f"{subcampaign_name}.{workflow_name}"] = (	1✔
62	workflow_config
63	)
64	else:
65	# Treat as regular workflow
66	workflows[workflow_key] = workflow_value	1✔
67
68	return workflows	1✔
69
70
71	def get_query_from_file(file_path: str) -> str:	1✔
72	"""
73	Extract a SQL query from a file.
74
75	Args:
76	file_path: The path to the file containing the SQL query.
77
78	Returns:
79	The SQL query as a string.
80	"""
81	query = "obs_id IN ("	1✔
82	with open(file_path, "r") as file:	1✔
83	obslist = file.readlines()	1✔
84	for obs_id in obslist:	1✔
85	obs_id = obs_id.strip()	1✔
86	query += f"'{obs_id}',"	1✔
87	query = query.rstrip(",")	1✔
88	query += ")"	1✔
89
90	return query	1✔

simonsobs / so_campaign_manager / 18854987527

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