"""User-facing automatic preparation with synthetic logs and disposable folders.""" import hashlib import io import json import os import tempfile import unittest import zipfile from contextlib import redirect_stderr, redirect_stdout from pathlib import Path from typing import Any from unittest.mock import patch import py7zr from netscaler_triage.cli import main from netscaler_triage.preparation import prepare_inputs LOG = ( '192.0.2.10 - - [01/Oct/2026:12:00:00 +0000] "GET /etc/passwd HTTP/1.1" 403 0 "-" "ordinary"\n' ) class EasyCliTests(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name).resolve() self.previous = Path.cwd() self.case = self.root / "case" self.case.mkdir() os.chdir(self.case) def tearDown(self) -> None: os.chdir(self.previous) self.temporary.cleanup() def invoke(self, arguments: list[str]) -> str: stdout, stderr = io.StringIO(), io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): result = main(arguments) self.assertEqual(result, 0, stdout.getvalue() + stderr.getvalue()) return stdout.getvalue() + stderr.getvalue() def events(self) -> list[dict[str, Any]]: files = list(self.root.rglob("findings.jsonl")) self.assertEqual(len(files), 1, files) return [json.loads(line) for line in files[0].read_text().splitlines()] def archive(self, name: str, log: str = LOG) -> Path: archive = self.case / name with zipfile.ZipFile(archive, "w") as handle: handle.writestr("var/log/httpaccess.log", log) return archive def encrypted_archive(self, password: str) -> Path: original = self.root / "fixture.log" original.write_text(LOG, encoding="utf-8") archive = self.case / "encrypted.7z" with py7zr.SevenZipFile(archive, "w", password=password, header_encryption=True) as handle: handle.write(original, "var/log/httpaccess.log") return archive def assert_password_absent(self, password: str, output: str) -> None: self.assertNotIn(password, output) for path in self.root.rglob("*"): if path.is_file() and path.suffix in {".json", ".jsonl", ".txt", ".csv"}: self.assertNotIn(password, path.read_text(encoding="utf-8"), str(path)) def test_no_arguments_uses_current_folder_and_skips_tool_directories(self) -> None: (self.case / "httpaccess.log").write_text(LOG, encoding="utf-8") (self.case / "pyproject.toml").write_text( '[project]\nname = "netscaler-triage"\n', encoding="utf-8" ) for name in ("src", ".runtime", ".venv", "results"): folder = self.case / name folder.mkdir() (folder / "unrelated.log").write_text(LOG * 3, encoding="utf-8") self.invoke([]) events = self.events() self.assertEqual(len(events), 1) self.assertEqual(events[0]["kind"], "http_access") def test_folder_without_subcommand(self) -> None: folder = self.case / "logs" folder.mkdir() (folder / "httpaccess.log").write_text(LOG, encoding="utf-8") self.invoke([str(folder)]) self.assertEqual(len(self.events()), 1) def test_explicit_scan_folder(self) -> None: folder = self.case / "logs" folder.mkdir() (folder / "httpaccess.log").write_text(LOG, encoding="utf-8") self.invoke(["scan", str(folder)]) self.assertEqual(len(self.events()), 1) def test_relative_output_preserves_preparation_manifest(self) -> None: folder = self.case / "logs" folder.mkdir() (folder / "httpaccess.log").write_text(LOG, encoding="utf-8") self.invoke(["scan", "logs", "--output", "results/new"]) output = self.case / "results/new" manifest = json.loads((output / "details/preparation_manifest.json").read_text()) self.assertTrue((output / "SUMMARY.txt").is_file()) for root in manifest["roots"]: self.assertTrue((output / root["path"]).is_dir()) self.assertEqual(len(self.events()), 1) def test_missing_common_drive_uses_current_output_parent(self) -> None: folder = self.case / "logs" folder.mkdir() (folder / "httpaccess.log").write_text(LOG, encoding="utf-8") with patch("netscaler_triage.cli.os.path.commonpath", side_effect=ValueError("drives")): self.invoke([str(folder)]) self.assertEqual(len(self.events()), 1) def test_advanced_preparation_limits_are_forwarded(self) -> None: folder = self.case / "logs" folder.mkdir() (folder / "httpaccess.log").write_text(LOG, encoding="utf-8") with patch("netscaler_triage.cli.prepare_inputs", wraps=prepare_inputs) as prepare: self.invoke( [str(folder), "--max-bytes", "10000", "--max-files", "9", "--max-depth", "0"] ) self.assertEqual(prepare.call_args.kwargs["max_bytes"], 10000) self.assertEqual(prepare.call_args.kwargs["max_files"], 9) self.assertEqual(prepare.call_args.kwargs["max_depth"], 0) self.assertEqual(len(self.events()), 1) def test_archive_is_prepared_automatically_without_changing_original(self) -> None: archive = self.archive("adc.zip") before = hashlib.sha256(archive.read_bytes()).digest() self.invoke([str(archive)]) self.assertEqual(hashlib.sha256(archive.read_bytes()).digest(), before) events = self.events() self.assertEqual(len(events), 1) self.assertEqual(events[0]["kind"], "http_access") def test_two_archives_keep_separate_appliance_identity(self) -> None: first = self.archive("adc-one.zip") second = self.archive("adc-two.zip") self.invoke([str(first), str(second)]) events = self.events() self.assertEqual(len(events), 2) self.assertEqual(len({event["appliance"] for event in events}), 2) def test_current_folder_discovers_multiple_archives(self) -> None: self.archive("adc-one.zip") self.archive("adc-two.zip") self.invoke([]) events = self.events() self.assertEqual(len(events), 2) self.assertEqual(len({event["appliance"] for event in events}), 2) def test_nested_archive_is_not_counted_twice(self) -> None: nested = io.BytesIO() with zipfile.ZipFile(nested, "w") as handle: handle.writestr("httpaccess.log", LOG) archive = self.case / "outer.zip" with zipfile.ZipFile(archive, "w") as handle: handle.writestr("inner.zip", nested.getvalue()) self.invoke([str(archive)]) self.assertEqual(len(self.events()), 1) def test_archive_password_is_not_written_to_output(self) -> None: password = "SYNTHETIC-password-SECRET-39217" archive = self.encrypted_archive(password) output = self.invoke([str(archive), "--password", password]) self.assertEqual(len(self.events()), 1) self.assert_password_absent(password, output) def test_password_flag_without_value_prompts_privately(self) -> None: password = "SYNTHETIC-prompt-SECRET-39218" archive = self.encrypted_archive(password) with patch("netscaler_triage.cli.getpass.getpass", return_value=password) as prompt: output = self.invoke([str(archive), "--password"]) prompt.assert_called_once() self.assertEqual(len(self.events()), 1) self.assert_password_absent(password, output) def test_password_option_alone_uses_current_folder(self) -> None: password = "SYNTHETIC-default-SECRET-39219" self.encrypted_archive(password) output = self.invoke(["--password", password]) self.assertEqual(len(self.events()), 1) self.assert_password_absent(password, output) def test_output_inside_original_input_is_rejected_before_changes(self) -> None: (self.case / "httpaccess.log").write_text(LOG, encoding="utf-8") before = { p.relative_to(self.root).as_posix(): p.read_bytes() if p.is_file() else None for p in self.root.rglob("*") } stdout, stderr = io.StringIO(), io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): result = main([str(self.case), "--output", str(self.case / "generated" / "out")]) self.assertNotEqual(result, 0) after = { p.relative_to(self.root).as_posix(): p.read_bytes() if p.is_file() else None for p in self.root.rglob("*") } self.assertEqual(before, after) def test_corrupt_archive_keeps_valid_results_and_reports_partial(self) -> None: (self.case / "httpaccess.log").write_text(LOG, encoding="utf-8") (self.case / "broken.zip").write_bytes(b"not a zip archive") stdout, stderr = io.StringIO(), io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): result = main([]) self.assertEqual(result, 2) self.assertEqual(len(self.events()), 1) summaries = list(self.root.rglob("summary.json")) self.assertEqual(len(summaries), 1) summary = json.loads(summaries[0].read_text()) self.assertEqual(summary["status"], "partial") self.assertEqual(summary["counts"]["preparation_errors"], 1) self.assertEqual(summary["preparation_errors"][0]["source"], "broken.zip") manifest = json.loads((summaries[0].parent / "preparation_manifest.json").read_text()) failures = [entry for entry in manifest["entries"] if entry["action"] == "error"] self.assertEqual(len(failures), 1) self.assertIn("Preparation incomplete", stderr.getvalue()) if __name__ == "__main__": unittest.main()