Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions src/telemetry_window_demo/io.py
Original file line number Diff line number Diff line change
Expand Up @@ -247,8 +247,7 @@ def _require_text_columns(


def write_table(frame: pd.DataFrame, path: str | Path) -> Path:
output_path = Path(path)
_ensure_output_directory(output_path.parent)
output_path = ensure_output_file_path(path)

export = frame.copy()
for column in export.columns:
Expand All @@ -264,8 +263,7 @@ def write_table(frame: pd.DataFrame, path: str | Path) -> Path:


def write_json(payload: dict[str, Any], path: str | Path) -> Path:
output_path = Path(path)
_ensure_output_directory(output_path.parent)
output_path = ensure_output_file_path(path)
output_path.write_text(
json.dumps(payload, indent=2) + "\n",
encoding="utf-8",
Expand All @@ -279,6 +277,14 @@ def ensure_output_directory(path: str | Path) -> Path:
return output_dir


def ensure_output_file_path(path: str | Path) -> Path:
output_path = Path(path)
if output_path.exists() and output_path.is_dir():
raise ValueError(f"Output file path is a directory: {output_path}")
_ensure_output_directory(output_path.parent)
return output_path


def _ensure_output_directory(output_dir: Path) -> None:
if output_dir.exists() and not output_dir.is_dir():
raise ValueError(f"Output directory path is not a directory: {output_dir}")
Expand Down
4 changes: 3 additions & 1 deletion src/telemetry_window_demo/visualize.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import matplotlib
import pandas as pd

from .io import ensure_output_directory
from .io import ensure_output_directory, ensure_output_file_path

matplotlib.use("Agg")
import matplotlib.pyplot as plt
Expand Down Expand Up @@ -51,6 +51,7 @@ def _plot_metric(
ylabel: str,
alerts: pd.DataFrame | None = None,
) -> Path:
output_path = ensure_output_file_path(output_path)
figure, axis = plt.subplots(figsize=(11, 4.5))
if features.empty:
axis.text(0.5, 0.5, "No feature windows generated", ha="center", va="center")
Expand Down Expand Up @@ -95,6 +96,7 @@ def _plot_metric(


def _plot_alert_timeline(alerts: pd.DataFrame, output_path: Path) -> Path:
output_path = ensure_output_file_path(output_path)
figure, axis = plt.subplots(figsize=(11, 4.5))
if alerts.empty:
axis.text(0.5, 0.5, "No alerts triggered", ha="center", va="center")
Expand Down
69 changes: 69 additions & 0 deletions tests/test_cli_errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,37 @@ def test_main_reports_file_output_dir_without_traceback(tmp_path, capsys) -> Non
assert "Traceback" not in stderr


def test_main_reports_directory_output_artifact_without_traceback(tmp_path, capsys) -> None:
input_path = tmp_path / "events.csv"
input_path.write_text(
"timestamp,event_type,source,target,status\n"
"2026-03-10T10:00:00Z,login_success,user_a,auth,ok\n",
encoding="utf-8",
)
output_dir = tmp_path / "processed"
(output_dir / "features.csv").mkdir(parents=True)
config_path = tmp_path / "directory-output-artifact.yaml"
config_path.write_text(
yaml.safe_dump(
{
"input_path": str(input_path),
"output_dir": str(output_dir),
}
),
encoding="utf-8",
)

with pytest.raises(SystemExit) as excinfo:
main(["run", "--config", str(config_path)])

assert excinfo.value.code == 1
stderr = capsys.readouterr().err
assert stderr.startswith("error: ")
assert "Output file path is a directory" in stderr
assert "features.csv" in stderr
assert "Traceback" not in stderr


def test_main_reports_bad_summarize_timestamp_column_without_traceback(
tmp_path,
capsys,
Expand Down Expand Up @@ -222,3 +253,41 @@ def test_main_reports_file_plot_output_dir_without_traceback(tmp_path, capsys) -
assert stderr.startswith("error: ")
assert "Output directory path is not a directory" in stderr
assert "Traceback" not in stderr


def test_main_reports_directory_plot_artifact_without_traceback(tmp_path, capsys) -> None:
features_path = tmp_path / "features.csv"
features_path.write_text(
"window_start,window_end,event_count,error_rate\n"
"2026-03-10T10:00:00Z,2026-03-10T10:01:00Z,10,0.25\n",
encoding="utf-8",
)
alerts_path = tmp_path / "alerts.csv"
alerts_path.write_text(
"alert_time,window_start,window_end,rule_name,severity\n"
"2026-03-10T10:01:00Z,2026-03-10T10:00:00Z,"
"2026-03-10T10:01:00Z,high_error_rate,medium\n",
encoding="utf-8",
)
output_dir = tmp_path / "plots"
(output_dir / "event_count_timeline.png").mkdir(parents=True)

with pytest.raises(SystemExit) as excinfo:
main(
[
"plot",
"--features",
str(features_path),
"--alerts",
str(alerts_path),
"--output-dir",
str(output_dir),
]
)

assert excinfo.value.code == 1
stderr = capsys.readouterr().err
assert stderr.startswith("error: ")
assert "Output file path is a directory" in stderr
assert "event_count_timeline.png" in stderr
assert "Traceback" not in stderr
Loading