diff --git a/README.md b/README.md index 27398f8..3d4cd99 100644 --- a/README.md +++ b/README.md @@ -106,6 +106,13 @@ finding. Do not send the key or an unredacted trace by email. ## Usage +Local `analyze()` accepts a nonempty Pisama native trace (`spans`), an ATIF +trajectory, or native span JSONL. OTLP `resourceSpans` exports are not a local +input format; use the hosted workflow above for those exports. Unsupported or +empty inputs raise an error rather than reporting a clean analysis. +JSONL may contain native span rows, or exactly one native trace envelope; mixing +envelopes with other rows is rejected rather than dropping later evidence. + ```python from pisama import analyze diff --git a/src/pisama/_loader.py b/src/pisama/_loader.py index 1fe9c7f..9feed00 100644 --- a/src/pisama/_loader.py +++ b/src/pisama/_loader.py @@ -12,6 +12,14 @@ def load_trace(input_data: Union[str, dict[str, Any], Trace]) -> Trace: + """Load supported trace input, rejecting empty traces before detection.""" + trace = _load_trace(input_data) + if not trace.spans: + raise ValueError("Trace contains no spans; no analysis was performed.") + return trace + + +def _load_trace(input_data: Union[str, dict[str, Any], Trace]) -> Trace: """Load a Trace from various input formats. Args: @@ -77,9 +85,46 @@ def _load_dict(data: dict[str, Any]) -> Trace: """Load either an ATIF trajectory or Pisama's native trace shape.""" if is_atif_trajectory(data): return trace_from_atif(data) + if "resourceSpans" in data: + raise ValueError( + "OTLP resourceSpans is not supported by local analyze(); " + "use hosted ingestion or provide an ATIF/native trace." + ) + if not isinstance(data.get("spans"), list): + raise ValueError("Expected an ATIF trajectory or a native trace with a spans list.") + for span in data["spans"]: + _validate_native_span(span) return Trace.from_dict(data) +def _validate_native_span(data: Any) -> None: + # IDs and all other individual fields are optional in native spans, but + # unrelated dictionaries must not silently become default/blank spans. + native_fields = { + "span_id", + "parent_id", + "trace_id", + "name", + "kind", + "platform", + "platform_metadata", + "start_time", + "end_time", + "status", + "error_message", + "attributes", + "events", + "input_data", + "output_data", + } + if ( + not isinstance(data, dict) + or not native_fields.intersection(data) + or {"resourceSpans", "spans", "steps"}.intersection(data) + ): + raise ValueError("Expected a native span object, not an empty object or trace/OTLP export.") + + def _load_jsonl(text: str) -> Trace: """Parse a JSONL file where each line is a span or event.""" lines = [line.strip() for line in text.splitlines() if line.strip()] @@ -87,16 +132,22 @@ def _load_jsonl(text: str) -> Trace: if not lines: raise ValueError("JSONL file is empty") - # If the first line parses as a full trace (has 'trace_id' + 'spans'), - # treat the file as a single-line trace dump. - first = json.loads(lines[0]) - if "trace_id" in first and "spans" in first: - return Trace.from_dict(first) + rows = [json.loads(line) for line in lines] + # A trace envelope is supported only as the sole row. Never discard + # subsequent evidence or silently flatten envelopes into blank spans. + if any(isinstance(row, dict) and "spans" in row for row in rows): + if len(rows) != 1: + raise ValueError( + "A JSONL trace envelope must be the only row; multiple rows cannot be merged." + ) + return _load_dict(rows[0]) # Otherwise, treat each line as a span dict and wrap them. from pisama_core.traces.models import Span - spans = [Span.from_dict(json.loads(line)) for line in lines] + for row in rows: + _validate_native_span(row) + spans = [Span.from_dict(row) for row in rows] trace = Trace() for span in spans: trace.add_span(span) diff --git a/tests/test_loader_fail_closed.py b/tests/test_loader_fail_closed.py new file mode 100644 index 0000000..a8287b6 --- /dev/null +++ b/tests/test_loader_fail_closed.py @@ -0,0 +1,103 @@ +"""Real parser and CLI checks: unsupported exports cannot look clean.""" + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest +from pisama_core.traces.models import Trace + +from pisama._loader import load_trace + + +@pytest.mark.parametrize("payload", [{"resourceSpans": []}, {}, {"spans": []}, {"spans": [None]}]) +def test_invalid_trace_rejected_from_dict_json_and_file(payload, tmp_path): + path = tmp_path / "trace.json" + path.write_text(json.dumps(payload)) + for value in (payload, json.dumps(payload), str(path)): + with pytest.raises(ValueError): + load_trace(value) + + +def test_empty_trace_object_rejected(): + with pytest.raises(ValueError, match="no spans"): + load_trace(Trace()) + + +@pytest.mark.parametrize("suffix", [".json", ".jsonl"]) +def test_otlp_cli_fails_without_clean_message(tmp_path, suffix): + path = tmp_path / ("trace" + suffix) + path.write_text( + json.dumps({"resourceSpans": [{"scopeSpans": [{"spans": [{"name": "synthetic"}]}]}]}) + ) + env = {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src")} + result = subprocess.run( + [sys.executable, "-c", "from pisama.cli.main import main; main()", "analyze", str(path)], + capture_output=True, + text=True, + env=env, + timeout=15, + ) + assert result.returncode == 1 + assert "OTLP" in result.stderr + assert "No issues detected" not in result.stdout + result.stderr + + +def test_native_span_without_id_remains_supported(tmp_path): + path = tmp_path / "trace.jsonl" + path.write_text(json.dumps({"name": "synthetic-span"})) + assert len(load_trace(str(path)).spans) == 1 + + +@pytest.mark.parametrize("span", [{}, {"unrelated": "value"}, {"resourceSpans": []}, None, []]) +def test_unsupported_span_shapes_rejected_in_native_and_jsonl(span, tmp_path): + with pytest.raises(ValueError): + load_trace({"spans": [span]}) + path = tmp_path / "trace.jsonl" + path.write_text(json.dumps(span)) + with pytest.raises(ValueError): + load_trace(str(path)) + + +@pytest.mark.parametrize("envelope_first", [True, False]) +@pytest.mark.parametrize("second_envelope", [True, False]) +def test_jsonl_envelope_cannot_discard_other_rows(tmp_path, envelope_first, second_envelope): + envelope = {"trace_id": "synthetic", "spans": [{"name": "first"}]} + evidence = {"name": "error-span", "status": "error", "error_message": "synthetic failure"} + other = {"trace_id": "second", "spans": [evidence]} if second_envelope else evidence + rows = [envelope, other] if envelope_first else [other, envelope] + path = tmp_path / "trace.jsonl" + path.write_text("\n".join(json.dumps(row) for row in rows)) + with pytest.raises(ValueError, match="only row"): + load_trace(str(path)) + + +def test_single_native_envelope_and_multiple_native_spans_supported(tmp_path): + spans = [{"name": "first"}, {"name": "second", "status": "error"}] + path = tmp_path / "trace.jsonl" + for payload in (json.dumps({"spans": spans}), "\n".join(map(json.dumps, spans))): + path.write_text(payload) + loaded = load_trace(str(path)) + assert len(loaded.spans) == 2 + assert loaded.spans[1].name == "second" + + +def test_cli_rejects_truncated_jsonl_analysis(tmp_path): + path = tmp_path / "trace.jsonl" + path.write_text( + json.dumps({"trace_id": "synthetic", "spans": [{"name": "first"}]}) + + "\n" + + json.dumps({"name": "later-error", "status": "error"}) + ) + result = subprocess.run( + [sys.executable, "-c", "from pisama.cli.main import main; main()", "analyze", str(path)], + capture_output=True, + text=True, + env={**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src")}, + timeout=15, + ) + assert result.returncode == 1 + assert "only row" in result.stderr + assert "No issues detected" not in result.stdout + result.stderr