Address review nits: timestamp, simulate, input filtering, docs

- default_timestamp no longer frozen at import time; computed per call
- simulate() no longer creates the output directory
- Input iteration skips non-.json/.jsn files and subdirectories
- class_ on compliance-type prop documented with a comment
- Added aws-config task section to cli.md and compliance_posture.md
- Added test for non-json/directory skipping

Addresses degenaro's Claude review on #2334.

Signed-off-by: Ahmed Hassan <th3reality72@gmail.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: AAH20

docs/tutorials/cli.md

@@ -1086,6 +1086,31 @@ Example output OSCAL Observations file contents (snippet):
 
 </details>
 
+## `trestle task aws-config-result-to-oscal-ar`
+
+The *trestle task aws-config-result-to-oscal-ar* command transforms AWS Config compliance evaluation results into OSCAL partial results `.json` files. Each input file is the JSON shape returned by `get-compliance-details-by-config-rule` / `get-compliance-details-by-resource` (`{"EvaluationResults": [EvaluationResult, ...]}`).
+
+Specify required config parameters for input and output directories. Optional `output-overwrite` controls whether existing output may be replaced. Optional `timestamp` is an ISO 8601 string that overrides the Result/Observation timestamps.
+
+<span style="color:green">
+Example command invocation:
+</span>
+
+`$TRESTLE_BASEDIR$ trestle task aws-config-result-to-oscal-ar -c /home/user/task.config`
+
+<span style="color:green">
+Example config:
+</span>
+
+```conf
+[task.aws-config-result-to-oscal-ar]
+input-dir = /home/user/git/compliance/aws-config/input
+output-dir = /home/user/git/compliance/oscal/output
+output-overwrite = true
+```
+
+Only `.json` / `.jsn` files in `input-dir` are processed. Nested directories and other extensions are skipped. `simulate` does not create the output directory.
+
 ## `trestle task tanium-result-to-oscal-ar`
 
 The *trestle task tanium-result-to-oscal-ar* command facilitates transformation of Tanuim reports, each

docs/tutorials/compliance_posture.md

@@ -38,7 +38,7 @@ The bad news is that a transformer to [OSCAL](https://pages.nist.gov/OSCAL) is n
 
 However, there is plenty of good news:
 
-- a transformer for your Cloud Service type may already exist, such as: [Tanium to OSCAL](../reference/API/trestle/tasks/tanium_result_to_oscal_ar.md), [OpenShift Compliance Operator to OSCAL](../reference/API/trestle/tasks/xccdf_result_to_oscal_ar.md)
+- a transformer for your Cloud Service type may already exist, such as: [Tanium to OSCAL](../reference/API/trestle/tasks/tanium_result_to_oscal_ar.md), [OpenShift Compliance Operator to OSCAL](../reference/API/trestle/tasks/xccdf_result_to_oscal_ar.md), [AWS Config to OSCAL](../reference/API/trestle/tasks/aws_config_result_to_oscal_ar.md)
 - once a transformer for a Cloud Service type has been written, it can be open-sourced/re-used
 - writing a transformer is fairly easy: just a few lines of Python code using [trestle](../index.md) as a foundation
 

tests/trestle/tasks/aws_config_result_to_oscal_ar_test.py

@@ -56,7 +56,7 @@ def test_simulate_does_not_write_output(self, tmp_path):
         task = AwsConfigResultToOscalAR(config)
         outcome = task.simulate()
         assert outcome == TaskOutcome.SIM_SUCCESS
-        assert not output_dir.exists() or not list(output_dir.iterdir())
+        assert not output_dir.exists()
 
     def test_execute_produces_valid_oscal_result(self, tmp_path):
         output_dir = tmp_path / 'out'
@@ -83,3 +83,19 @@ def test_execute_respects_output_overwrite_false(self, tmp_path):
         config2['output-overwrite'] = 'false'
         task2 = AwsConfigResultToOscalAR(config2)
         assert task2.execute() == TaskOutcome.FAILURE
+
+    def test_execute_skips_non_json_and_directories(self, tmp_path):
+        input_dir = tmp_path / 'in'
+        input_dir.mkdir()
+        (input_dir / 'nested').mkdir()
+        (input_dir / 'readme.txt').write_text('not json', encoding='utf-8')
+        sample = test_data_dir / 'aws-config-sample.json'
+        (input_dir / 'aws-config-sample.json').write_text(sample.read_text(encoding='utf-8'), encoding='utf-8')
+
+        output_dir = tmp_path / 'out'
+        config = _build_config(input_dir, output_dir)
+        task = AwsConfigResultToOscalAR(config)
+        assert task.execute() == TaskOutcome.SUCCESS
+        produced = list(output_dir.glob('*.oscal.json'))
+        assert len(produced) == 1
+        assert produced[0].name == 'aws-config-sample.oscal.json'

trestle/tasks/aws_config_result_to_oscal_ar.py

@@ -130,10 +130,11 @@ def _transform_work(self) -> TaskOutcome:
             except Exception:
                 logger.warning('config invalid "timestamp"')
                 return TaskOutcome(mode + 'failure')
-        # ensure output dir exists
-        opth.mkdir(exist_ok=True, parents=True)
-        # process
+        if not self._simulate:
+            opth.mkdir(exist_ok=True, parents=True)
         for ifile in sorted(ipth.iterdir()):
+            if not ifile.is_file() or ifile.suffix not in ['.json', '.jsn']:
+                continue
             blob = self._read_file(ifile)
             transformer = AwsConfigResultToOscalARTransformer()
             results = transformer.transform(blob)

trestle/transforms/implementations/aws_config.py

@@ -88,11 +88,9 @@ def transform(self, blob: str) -> Results:
 class _OscalResultsFactory:
     """Build OSCAL entities from AWS Config EvaluationResult objects."""
 
-    default_timestamp = ResultsTransformer.get_timestamp()
-
-    def __init__(self, timestamp: str = default_timestamp) -> None:
+    def __init__(self, timestamp: str | None = None) -> None:
         """Initialize."""
-        self._timestamp = timestamp
+        self._timestamp = timestamp if timestamp is not None else ResultsTransformer.get_timestamp()
         self._observation_list: List[Observation] = []
         self._inventory_map: Dict[str, InventoryItem] = {}
         self._ns = AnyUrl('https://oscal-compass.github.io/compliance-trestle/schemas/oscal/ar/aws-config')
@@ -187,6 +185,10 @@ def _observation_properties(self, evaluation_result: Dict[str, Any]) -> List[Pro
             ('result-token', evaluation_result.get('ResultToken')),
         ):
             if value is not None:
+                # Only compliance-type is classed for downstream filtering,
+                # matching osco.py's scc_result pattern. Prop names stay
+                # hyphenated (OSCAL); class_ uses the underscored identifier
+                # style of osco's scc_* values.
                 class_ = 'aws_config_compliance' if name == 'compliance-type' else None
                 if class_:
                     props.append(Property.model_construct(name=name, value=str(value), ns=self._ns, class_=class_))