111 lines
5.5 KiB
Python
111 lines
5.5 KiB
Python
"""Working field policy for the campaign-importer pilot, independent of exports."""
|
|
|
|
import copy
|
|
import math
|
|
|
|
|
|
ANALYSIS_ALLOWLIST = frozenset({
|
|
"campaign", "campaign_family", "suite", "class", "case_id", "description", "case_type",
|
|
"verification_method", "verifies", "traced_to", "functional_area", "functional_group",
|
|
"functional_group_name", "swci", "operating_condition", "preconditions", "success_criteria",
|
|
"witness", "atp", "qtp", "fqt", "etr_suite",
|
|
})
|
|
EXPORT_ALLOWLIST = ANALYSIS_ALLOWLIST - {
|
|
"description", "operating_condition", "preconditions", "success_criteria",
|
|
}
|
|
PROFILE_FIELDS = frozenset({
|
|
"campaign", "suite", "case_id", "description", "case_type", "verification_method",
|
|
"swci", "operating_condition", "preconditions", "success_criteria",
|
|
})
|
|
GROUP_FIELDS = PROFILE_FIELDS | {
|
|
"campaign_family", "class", "verifies", "traced_to", "functional_area",
|
|
"functional_group", "functional_group_name",
|
|
}
|
|
EXPORT_SOURCE_ROWS = False
|
|
|
|
|
|
def _flat_value(value):
|
|
"""Accept normalized spreadsheet scalars/lists, never nested raw dictionaries."""
|
|
if value is None or type(value) in (str, bool, int):
|
|
return value
|
|
if type(value) is float and math.isfinite(value):
|
|
return value
|
|
if isinstance(value, (list, tuple)) and all(not isinstance(v, (dict, list, tuple)) for v in value):
|
|
return [_flat_value(item) for item in value]
|
|
raise ValueError("normalize field values to JSON scalars before invoking inference")
|
|
|
|
|
|
def identity(record):
|
|
"""Require exact source identifiers instead of normalizing or guessing them."""
|
|
result = tuple(record.get(key) for key in ("campaign", "suite", "case_id"))
|
|
if any(not isinstance(value, str) or not value for value in result):
|
|
raise ValueError("campaign, suite and case_id must be nonempty source identifiers")
|
|
return result
|
|
|
|
|
|
def select_fa01(records):
|
|
"""Filter the laptop's source records by exact campaign identity before networking.
|
|
|
|
The adapter requires the actual identifier FA01. Any mapping from a composite
|
|
workbook label must be inspected in the application before wiring this call.
|
|
"""
|
|
selected = [copy.deepcopy(record) for record in records if record.get("campaign") == "FA01"]
|
|
keys = [identity(record) for record in selected]
|
|
if not selected or len(keys) != len(set(keys)):
|
|
raise ValueError("no exact FA01 campaign match, or duplicate source case identity")
|
|
return selected
|
|
|
|
|
|
def analysis_view(record, operation):
|
|
"""Minimize inference fields by operation; witness/reporting flags are omitted."""
|
|
if operation not in ("profile", "group"):
|
|
raise ValueError("unknown analysis operation")
|
|
identity(record)
|
|
wanted = PROFILE_FIELDS if operation == "profile" else GROUP_FIELDS
|
|
return {key: _flat_value(record[key]) for key in sorted(wanted & ANALYSIS_ALLOWLIST) if key in record}
|
|
|
|
|
|
def export_view(record, *, enabled_fields=EXPORT_ALLOWLIST, export_source_rows=EXPORT_SOURCE_ROWS):
|
|
"""Build a local export preview from originals; caller flags may only narrow access."""
|
|
identity(record)
|
|
fields = EXPORT_ALLOWLIST & set(enabled_fields)
|
|
result = {key: _flat_value(record[key]) for key in sorted(fields) if key in record}
|
|
if export_source_rows and "source_row" in record:
|
|
value = record["source_row"]
|
|
if type(value) is not int or value < 1:
|
|
raise ValueError("source_row must be a positive integer")
|
|
result["source_row"] = value
|
|
return result
|
|
|
|
|
|
def safe_family_previews(grouping, original_records, *, enabled_fields=EXPORT_ALLOWLIST,
|
|
export_source_rows=EXPORT_SOURCE_ROWS):
|
|
"""Use allowed assignments and original fields, never model-generated export text.
|
|
|
|
This returns local preview data only. It cannot contact ClickUp and carries no
|
|
attachments or model-generated titles, descriptions or objective summaries.
|
|
"""
|
|
from collections import Counter
|
|
|
|
if not {"campaign", "suite", "case_id"} <= (EXPORT_ALLOWLIST & set(enabled_fields)):
|
|
raise ValueError("traceable family previews require export-enabled source identifiers")
|
|
owners = {identity(record)[:2] for record in original_records}
|
|
if owners != {(grouping.get("campaign_id"), grouping.get("suite_id"))}:
|
|
raise ValueError("export preview must use one original campaign/suite")
|
|
records = {identity(record)[2]: record for record in original_records}
|
|
members = [case_id for family in grouping["families"] for case_id in family["member_case_ids"]]
|
|
if len(records) != len(original_records) or Counter(members) != Counter(records.keys()):
|
|
raise ValueError("family assignments must partition the original cases exactly")
|
|
previews = []
|
|
for index, family in enumerate(grouping["families"], 1):
|
|
objectives = [{"title": "Case objective " + case_id,
|
|
"fields": export_view(records[case_id], enabled_fields=enabled_fields,
|
|
export_source_rows=export_source_rows)}
|
|
for case_id in family["member_case_ids"]]
|
|
previews.append({"title": f"Implementation family {index:03d}",
|
|
"campaign": grouping["campaign_id"], "suite": grouping["suite_id"],
|
|
"member_case_ids": list(family["member_case_ids"]),
|
|
"description": "Implementation family with individually traceable case objectives.",
|
|
"objectives": objectives, "attachments": [], "publishing_enabled": False})
|
|
return previews
|