codeflash-agent/evals/templates/crossdomain-easy/tests/test_analyzer.py
Kevin Turcios 33faedf427
Add Unstructured report, rewrite statusline, format evals/scripts (#20)
* Add Unstructured engagement report as uv workspace member

Three-tier Plotly Dash app (Executive Brief, Engineering Team, Full
Detail) with data in JSON, theme constants in theme.py, and Dash
production improvements (Google Fonts, clientside callbacks, meta tags).

Also: add .playwright-mcp/ to .gitignore, add reports/* ruff overrides,
remove tracked .codeflash/observability/read-tracker.

* Rewrite statusline to derive context from git state

Detects active area from changed files (reports, packages, plugin,
.codeflash, case-studies, evals), falls back to branch name convention
(perf/*, feat/*, fix/*), shows dirty indicator. Uses whoami for
cross-platform user detection.

* Add pre-push lint rule to commit guidelines

* Exclude .codeflash/ from ruff linting

Benchmark and profiling scripts in .codeflash/ are scratch work, not
package source. Excluding them prevents CI failures from ad-hoc scripts.

* Run ruff format across packages, scripts, evals, and plugin refs

* Fix github-app async test failures in CI

Add asyncio_mode = "auto" to root pytest config so async tests
are detected when running from the repo root via uv run pytest packages/.
2026-04-15 03:06:16 -05:00

31 lines
1.1 KiB
Python

from log_analyzer.analyzer import analyze_logs
def test_analyze_logs_basic():
entries = [
{"level": "ERROR", "source": "auth", "message": "login failed"},
{"level": "INFO", "source": "auth", "message": "login succeeded"},
{"level": "ERROR", "source": "db", "message": "connection timeout"},
]
result = analyze_logs(entries)
assert "frequencies" in result
assert "anomalies" in result
assert len(result["anomalies"]) > 0
def test_analyze_large_batch():
"""Realistic batch — this is the one that uses too much memory."""
sources = [f"service-{i}" for i in range(50)]
levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
entries = []
for i in range(10_000):
entries.append(
{
"level": levels[(i // len(sources)) % len(levels)],
"source": sources[i % len(sources)],
"message": f"Event {i}: something happened in the system",
}
)
result = analyze_logs(entries)
assert len(result["frequencies"]) == len(sources) * len(levels)
assert len(result["anomalies"]) > 0