58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from pathlib import Path
|
|
import tempfile
|
|
|
|
import log_preanalysis as lp
|
|
|
|
|
|
def test_detect_job_and_date() -> None:
|
|
query = "Разбери заказ 17876 за 2026-08-06: расхождение, Error64, промежутки, частота"
|
|
assert lp.looks_like_log_job(query, ["line_u_codes.log.1"])
|
|
assert lp.detect_job_id(query) == "17876"
|
|
assert lp.detect_shift_date(query) == "2026-08-06"
|
|
assert lp.detect_shift_date("смена 06.08.2026") == "2026-08-06"
|
|
|
|
|
|
def test_stream_scan(tmp_path: Path) -> None:
|
|
log = tmp_path / "line_u_codes.log.1"
|
|
lines = [
|
|
"2026-08-06 08:00:00 [Scanner] job=17876 code=010460406000000021ABC1",
|
|
"2026-08-06 08:00:01 [PLC] job=17876 Error 12",
|
|
"2026-08-06 08:05:00 [Scanner] job=17876 Error64 timeout",
|
|
"2026-08-06 08:05:01 [PLC] Error65",
|
|
"2026-08-06 10:00:00 [Scanner] last line",
|
|
]
|
|
log.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
percents: list[int] = []
|
|
|
|
def on_progress(info: dict) -> None:
|
|
if info.get("percent") is not None:
|
|
percents.append(int(info["percent"]))
|
|
|
|
report = lp.preanalyze_logs(
|
|
[(log, "line_u_codes.log.1")],
|
|
"заказ 17876 2026-08-06 Error64 промежутки частота",
|
|
on_progress=on_progress,
|
|
)
|
|
assert report["job_id"] == "17876"
|
|
item = report["files"][0]
|
|
assert item["lines"] == 5
|
|
assert item["error64"]
|
|
assert item["error65"]
|
|
assert item["gaps"]
|
|
assert item["gaps"][0]["seconds"] >= 30
|
|
md = report["markdown"]
|
|
assert "PREANALYSIS" in md
|
|
assert "Error64" in md
|
|
text, encoding = lp.excerpt_lines(log, 2, 3)
|
|
assert encoding == "utf-8"
|
|
assert "2\t" in text and "Error 12" in text
|
|
assert percents
|
|
assert percents[-1] == 100
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_detect_job_and_date()
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
test_stream_scan(Path(folder))
|
|
print("ok")
|