from datetime import UTC, datetime, timedelta from beaver_agent.hands.komodo import Komodo from beaver_agent.hands.komodo_logs import ( KEEP, SELF_MARK, Seen, StackScan, Sweep, brief, daily, digest, grouped, read_stack, ) from beaver_agent.jobs.logs import sweep as sweep_fleet NOW = datetime(2026, 9, 20, 18, 0, tzinfo=UTC) def log(*lines): return {"stdout": "\n".join(lines), "stderr": ""} def line(service, at, text): return f"{service} | {at}Z {text}" def a_stack(name, server="dell-home"): return { "id": name, "name": name, "info": {"state": "running", "server_name": server}, } def fleet(answers, calls=None): """Komodo, отвечающий заготовками; calls копит запросы к нему.""" class Fake(Komodo): async def read(self, request, params): if calls is not None: calls.append((request, params)) answer = answers.get(request, []) return answer(params) if callable(answer) else answer return Fake(url="u", key="k", secret="s") def swept(scans, **kw): kw.setdefault("in_scope", len(scans)) kw.setdefault("fleet", len(scans)) return Sweep(at=kw.pop("at", NOW), scans=tuple(scans), **kw) KNOWN = ("beaver-agent", "gitea", "democalc", "mix", "healthbot") def seen(tmp_path, *, fresh=False, known=KNOWN): """Состояние с уже снятой базовой линией: проход не первый.""" state = Seen.load(tmp_path / "логи.json") if not fresh: state.fresh = False state.baselined = set(known) return state def test_a_line_gives_its_service_its_time_and_its_signal(): scan = read_stack( "beaver-agent", log( line("beaver-gateway", "2026-09-20T17:40:01.123456789", "ERROR boom"), line("beaver-gateway", "2026-09-20T17:41:02.000000000", "ERROR boom"), line("beaver-obsidian", "2026-09-20T17:42:03.000000000", "Fully synced"), ), ) (hit,) = scan.hits assert hit.service == "beaver-gateway" assert hit.signal == "ошибка" assert hit.count == 2 assert hit.at == datetime(2026, 9, 20, 17, 41, 2, tzinfo=UTC) assert scan.first == datetime(2026, 9, 20, 17, 40, 1, 123456, tzinfo=UTC) assert scan.last == datetime(2026, 9, 20, 17, 42, 3, tzinfo=UTC) def test_the_same_trouble_with_other_numbers_is_the_same_trouble(): scan = read_stack( "gitea", log( line("gitea", "2026-09-20T17:40:01.0", "ERROR request 8871 failed"), line("gitea", "2026-09-20T17:40:09.0", "ERROR request 9134 failed"), ), ) (hit,) = scan.hits assert hit.count == 2 def test_every_named_signal_is_caught(): texts = ( "CRITICAL disk is gone", "panic: runtime error", "container OOMKilled", "Traceback (most recent call last):", "service exited with code 137", "beaver-gateway Restarting (1) 3 seconds ago", "container is unhealthy", "ERROR plain", ) scan = read_stack( "mix", log(*(line("svc", f"2026-09-20T17:4{i}:00.0", t) for i, t in enumerate(texts))), ) assert {h.signal for h in scan.hits} == { "критично", "паника", "oom", "traceback", "падение", "рестарт-цикл", "healthcheck", "ошибка", } assert "exited with code 0" not in {h.sample for h in scan.hits} def test_a_clean_exit_and_our_own_inject_are_not_trouble(): scan = read_stack( "beaver-agent", log( line("svc", "2026-09-20T17:40:00.0", "worker exited with code 0"), line("beaver-gateway", "2026-09-20T17:41:00.0", f"{SELF_MARK}: ERROR boom"), ), ) assert scan.hits == () def test_komodos_own_error_is_not_a_clean_stack(): scan = read_stack("gitea", {"error": "periphery unreachable"}) assert not scan.ok assert scan.error == "periphery unreachable" assert Sweep(at=NOW, scans=(scan,), in_scope=1, fleet=1).trouble == ( "gitea: periphery unreachable", ) def test_the_first_pass_only_takes_the_baseline(tmp_path): state = seen(tmp_path, fresh=True) scan = read_stack( "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR старая беда")) ) report = digest(state, swept([scan])) assert report.cold and report.baseline == 1 assert report.new == () text = brief(report) assert "базовую линию" in text assert "прочитано 1 стеков из 1" in text def test_only_the_new_line_wakes_the_master(tmp_path): old = line("svc", "2026-09-20T17:40:00.0", "ERROR молотит третьи сутки") state = seen(tmp_path, fresh=True) first = digest(state, swept([read_stack("beaver-agent", log(old))])) state.mark_told(first) state.save() fresh_line = line("svc", "2026-09-20T18:50:00.0", "CRITICAL диск кончился") state = Seen.load(tmp_path / "логи.json") report = digest( state, swept( [read_stack("beaver-agent", log(old, old, fresh_line))], at=NOW + timedelta(hours=3), ), ) (new,) = report.new assert new.signal == "критично" text = brief(report) assert "диск кончился" in text assert "молотит третьи сутки" not in text def test_a_repeat_carries_how_many_times(tmp_path): noisy = [ line("svc", f"2026-09-20T17:4{i}:00.0", "ERROR раз за разом") for i in range(5) ] report = digest(seen(tmp_path), swept([read_stack("beaver-agent", log(*noisy))])) assert "×5" in brief(report) def test_a_trouble_that_stopped_is_dated_by_the_event_not_by_the_look(tmp_path): was = line("svc", "2026-09-20T12:03:00.0", "ERROR было и прошло") state = seen(tmp_path) state.mark_told(digest(state, swept([read_stack("beaver-agent", log(was))]))) later = NOW + timedelta(hours=6) report = digest(state, swept([read_stack("beaver-agent", log())], at=later)) (quiet,) = report.quiet assert quiet.last == datetime(2026, 9, 20, 12, 3, tzinfo=UTC) text = brief(report) assert "последнее в 20.09 12:03" in text assert f"{later:%H:%M}" not in text.split("\n", 1)[1] def test_a_stack_that_did_not_answer_is_not_a_stack_that_went_quiet(tmp_path): was = line("svc", "2026-09-20T12:03:00.0", "ERROR было") state = seen(tmp_path) state.mark_told(digest(state, swept([read_stack("beaver-agent", log(was))]))) report = digest( state, swept( [StackScan(stack="beaver-agent", error="timeout")], at=NOW + timedelta(hours=3), ), ) assert report.quiet == () assert report.trouble == ("beaver-agent: timeout",) assert "beaver-agent: timeout" in brief(report) def test_an_undelivered_inject_leaves_the_trouble_new(tmp_path): scan = read_stack( "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) ) state = seen(tmp_path) first = digest(state, swept([scan])) assert len(first.new) == 1 state.save() # инжект не доехал: mark_told не звали state = Seen.load(state.path) again = digest( state, swept([read_stack("beaver-agent", log())], at=NOW + timedelta(hours=3)) ) assert [e.sample for e in again.new] == [e.sample for e in first.new] assert again.quiet == () def test_the_state_survives_a_restart(tmp_path): scan = read_stack( "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) ) state = seen(tmp_path) state.mark_told(digest(state, swept([scan]))) state.save() state = Seen.load(state.path) assert not state.fresh report = digest(state, swept([scan], at=NOW + timedelta(hours=3))) assert report.new == () and report.quiet == () def test_an_old_line_still_lying_in_the_tail_is_not_new_again(tmp_path): # Тихий стек держит августовскую строку в хвосте месяцами: она не новая. scan = read_stack( "democalc", log(line("pg-1", "2026-08-22T15:49:51.0", "ERROR: relation")) ) state = seen(tmp_path) state.mark_told(digest(state, swept([scan]))) state.save() state = Seen.load(state.path) assert digest(state, swept([scan], at=NOW + timedelta(hours=3))).new == () def test_a_trouble_forgotten_by_time_comes_back_as_new(tmp_path): scan = read_stack( "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) ) state = seen(tmp_path) state.mark_told(digest(state, swept([scan]))) late = NOW + KEEP + timedelta(days=1) report = digest(state, swept([scan], at=late)) assert len(report.new) == 1 def test_a_silent_pass_is_not_worth_the_master(tmp_path): state = seen(tmp_path) assert not digest(state, swept([read_stack("beaver-agent", log())])).worth_telling def test_the_scope_says_how_much_was_looked_at(tmp_path): scans = [ read_stack( "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) ), StackScan(stack="gitea", error="timeout"), ] report = digest(seen(tmp_path), swept(scans, in_scope=2, fleet=53)) head = brief(report).splitlines()[1] assert head.startswith( "Область: dell-home, personal-main - прочитано 1 стеков из 2" ) assert "во флоте 53" in head assert "хвост 2000 строк на стек: строки с 20.09 17:40 по 20.09 17:40" in head assert "самый короткий хвост у beaver-agent - только с 20.09 17:40" in head def test_a_tail_shorter_than_the_pause_between_passes_is_said_out_loud(tmp_path): state = seen(tmp_path) state.last_pass = NOW - timedelta(hours=3) scan = read_stack( "beaver-agent", log(line("svc", "2026-09-20T17:59:00.0", "ERROR из последней минуты")), ) report = digest(state, swept([scan])) assert report.gaps == ("beaver-agent",) assert "часть строк прошла мимо" in brief(report) def test_a_silent_komodo_says_so_instead_of_reporting_clean(tmp_path): down = Sweep(at=NOW, unreachable="ClientError: connection refused") report = digest(seen(tmp_path), down) assert report.worth_telling text = brief(report) assert "Komodo не ответил" in text assert "Нового в этой области нет" not in text def test_a_lasting_outage_is_told_once(tmp_path): down = Sweep(at=NOW, unreachable="ClientError: connection refused") state = seen(tmp_path) state.mark_told(digest(state, down)) assert digest(state, down).trouble == () async def test_the_sweep_lists_every_page_and_skips_other_peoples_machines(): pages = [ [a_stack(f"s{i}") for i in range(500)], [a_stack("gitea", "personal-main"), a_stack("veris-runner", "veris-main")], ] calls = [] komodo = fleet( { "ListStacks": lambda p: pages[p["page"]], "GetStacksSummary": {"total": 502}, "GetStackLog": log(), }, calls, ) out = await sweep_fleet(komodo, now=NOW) asked = [p["stack"] for r, p in calls if r == "GetStackLog"] assert "veris-runner" not in asked assert "gitea" in asked and "s0" in asked assert out.fleet == 502 and out.in_scope == 501 assert out.trouble == () assert all(p.get("timestamps") for r, p in calls if r == "GetStackLog") async def test_the_sweep_shouts_when_the_list_and_the_summary_disagree(): komodo = fleet( { "ListStacks": [a_stack("beaver-agent")], "GetStacksSummary": {"total": 53}, "GetStackLog": log(), } ) out = await sweep_fleet(komodo, now=NOW) assert out.trouble == ( "листание дало 1 стеков, а GetStacksSummary - 53: список неполон", ) async def test_a_stack_that_throws_does_not_end_the_sweep(): def logs(params): if params["stack"] == "gitea": msg = "connection reset" raise RuntimeError(msg) return log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) komodo = fleet( { "ListStacks": [a_stack("beaver-agent"), a_stack("gitea", "personal-main")], "GetStacksSummary": {"total": 2}, "GetStackLog": logs, } ) out = await sweep_fleet(komodo, now=NOW) assert len(out.read) == 1 assert out.trouble == ("gitea: RuntimeError: connection reset",) async def test_a_dead_komodo_makes_a_sweep_that_admits_it(): class Dead(Komodo): async def read(self, request, params): msg = "connection refused" raise OSError(msg) out = await sweep_fleet(Dead(url="u", key="k", secret="s"), now=NOW) assert out.read == () assert out.unreachable == "OSError: connection refused" assert out.trouble == ("Komodo не ответил: OSError: connection refused",) def test_a_line_without_a_stamp_is_not_a_find(): # Голову хвоста Komodo обрезает по месту: полстроки - не новая беда. scan = read_stack( "democalc", log( "49.260 UTC [8772] ERROR: relation does not exist", line("postgres-1", "2026-09-20T17:40:00.0", "ERROR: настоящая"), ), ) assert [h.sample for h in scan.hits] == ["ERROR: настоящая"] assert scan.unparsed == 1 and scan.lines == 1 def test_many_unreadable_lines_are_said_out_loud(tmp_path): scan = StackScan(stack="beaver-agent", lines=5, unparsed=40) head = brief(digest(seen(tmp_path), swept([scan]))).splitlines()[1] assert "строк без метки времени пропущено 40" in head def test_an_empty_line_is_a_line_not_a_blind_spot(): scan = read_stack("dell-home-dawarich", log("app-1 | 2026-09-17T14:37:38.510Z")) assert scan.unparsed == 0 and scan.lines == 1 and scan.hits == () def test_a_stack_read_for_the_first_time_does_not_dump_its_past(tmp_path): state = seen(tmp_path) scan = read_stack( "lodge", log(line("svc", "2026-09-19T10:00:00.0", "ERROR старое, до нас")) ) report = digest(state, swept([scan])) assert report.new == () assert report.baselined == ("lodge",) and report.baseline == 1 assert "Впервые прочитаны" in brief(report) later = read_stack( "lodge", log( line("svc", "2026-09-19T10:00:00.0", "ERROR старое, до нас"), line("svc", "2026-09-20T19:00:00.0", "ERROR уже при нас"), ), ) after = digest(state, swept([later], at=NOW + timedelta(hours=3))) assert [e.sample for e in after.new] == ["ERROR уже при нас"] def test_a_first_pass_into_a_dead_komodo_does_not_burn_the_baseline(tmp_path): state = seen(tmp_path, fresh=True) down = Sweep(at=NOW, unreachable="OSError: connection refused") report = digest(state, down) assert not report.cold and state.fresh assert report.trouble == ("Komodo не ответил: OSError: connection refused",) scan = read_stack( "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) ) later = digest(state, swept([scan], at=NOW + timedelta(hours=3))) assert later.cold and later.new == () def test_a_named_secret_in_the_line_does_not_ride_into_the_chat(): scan = read_stack( "gitea", log(line("svc", "2026-09-20T17:40:00.0", "ERROR auth failed token=ghp_ZZ123")), ) (hit,) = scan.hits assert "ghp_ZZ123" not in hit.sample assert hit.sample.startswith("ERROR auth failed token=") def test_a_bare_blob_stays_readable(): # Размен с открытыми глазами: id задач, сессий и коммитов - это сигнал, # и замазывать их ради редкого безымянного ключа дороже, чем оставить. scan = read_stack( "komodo", log(line("svc", "2026-09-20T17:40:00.0", "ERROR task kJ8sdfLKJ23kjhsdf98234x")), ) (hit,) = scan.hits assert "kJ8sdfLKJ23kjhsdf98234x" in hit.sample def stamps(day, hour, count, start=0): return [ f"2026-09-{day:02d}T{hour:02d}:{m:02d}:00.0" for m in range(start, start + count) ] def burst(stack, times, text="ERROR всё та же беда", service="svc"): """Хвост стека, где одна и та же беда повторилась в перечисленные минуты.""" return read_stack(stack, log(*(line(service, t, text) for t in times))) def at(day, hour, minute=0): return datetime(2026, 9, day, hour, minute, tzinfo=UTC) def told(state, report): state.mark_told(report) return report def test_a_continuing_trouble_is_told_once_and_then_keeps_quiet(tmp_path): state = seen(tmp_path) first = told( state, digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, 10))) ) assert len(first.new) == 1 # Хвост перечитан: те же строки плюс свежие - беда идёт, но новостью не стала. later = digest( state, swept([burst("gitea", stamps(20, 9, 3) + stamps(20, 12, 4))], at=at(20, 13)), ) assert later.new == () and later.surge == () and later.quiet == () assert not later.worth_telling assert later.ongoing == () # за прошлые сутки её ещё не было def test_the_reread_tail_does_not_inflate_the_count(tmp_path): state = seen(tmp_path) told(state, digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, 10)))) for hour in (13, 16, 19): digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, hour))) (entry,) = state.entries.values() assert entry.count == 3 # три повтора, а не три на каждый из четырёх проходов assert entry.passes == 1 def test_a_continuing_trouble_comes_back_in_the_daily_summary(tmp_path): state = seen(tmp_path) told(state, digest(state, swept([burst("gitea", stamps(20, 9, 5))], at=at(20, 10)))) # Первый проход следующих суток - 01:40, он же несёт сводку. night = digest( state, swept([burst("gitea", stamps(20, 9, 5) + stamps(21, 1, 2))], at=at(21, 1, 40)), ) assert night.summary and len(night.ongoing) == 1 assert night.new == () and not night.worth_telling text = daily(night, UTC) assert "длящееся, сводка за 2026-09-21" in text assert "gitea/svc ошибка, с 20.09 09:00, за сутки 5, первые сутки" in text assert "ERROR" not in text # образцов строк в сводке нет намеренно told(state, night) state.mark_daily(night) # Тот же день, следующий проход: сводку уже отдали. again = digest( state, swept([burst("gitea", stamps(20, 9, 5) + stamps(21, 1, 2))], at=at(21, 4, 40)), ) assert again.ongoing and not again.summary def test_the_summary_says_which_way_it_is_going(tmp_path): state = seen(tmp_path) told(state, digest(state, swept([burst("gitea", stamps(20, 9, 2))], at=at(20, 10)))) day_two = swept( [burst("gitea", stamps(20, 9, 2) + stamps(21, 9, 8))], at=at(21, 10) ) told(state, digest(state, day_two)) third = digest( state, swept( [burst("gitea", stamps(20, 9, 2) + stamps(21, 9, 8) + stamps(22, 1, 1))], at=at(22, 1, 40), ), ) assert "за сутки 8, растёт (было 2)" in daily(third, UTC) def test_a_surge_of_the_familiar_breaks_the_silence(tmp_path): state = seen(tmp_path) state_report = digest( state, swept([burst("gitea", stamps(20, 9, 1))], at=at(20, 10)) ) told(state, state_report) for hour, minute in ((13, 12), (16, 15)): digest( state, swept([burst("gitea", stamps(20, hour - 1, 1, minute))], at=at(20, hour)), ) (entry,) = state.entries.values() assert entry.passes == 3 and entry.norm == 1.0 spike = digest(state, swept([burst("gitea", stamps(20, 18, 12))], at=at(20, 19))) (item,) = spike.surge assert item.gain == 12 and item.norm == 1.0 assert spike.new == () and spike.worth_telling text = brief(spike) assert "Всплеск знакомого (1)" in text assert "×12 за проход при норме ×1.0" in text assert "Нового в этой области нет" not in text # Второй такой же проход внутри выдержки молчит: это уже не новость. told(state, spike) more = digest(state, swept([burst("gitea", stamps(20, 21, 12))], at=at(20, 22))) assert more.surge == () and not more.worth_telling def test_a_small_jump_is_not_a_surge(tmp_path): state = seen(tmp_path) told(state, digest(state, swept([burst("gitea", stamps(20, 9, 1))], at=at(20, 10)))) for hour in (13, 16): digest( state, swept([burst("gitea", stamps(20, hour - 1, 1, hour))], at=at(20, hour)), ) # Вчетверо больше, но всего четыре штуки: пол в SURGE_MIN держит тишину. small = digest(state, swept([burst("gitea", stamps(20, 18, 4))], at=at(20, 19))) assert small.surge == () and not small.worth_telling def test_a_trouble_gone_silent_is_reported_even_while_its_line_lingers(tmp_path): state = seen(tmp_path) tail = burst("gitea", stamps(20, 9, 2)) told(state, digest(state, swept([tail], at=at(20, 10)))) # Сутки та же строка лежит в хвосте, новых событий нет. assert not digest(state, swept([tail], at=at(20, 22))).quiet stopped = digest(state, swept([tail], at=at(21, 12))) (gone,) = stopped.quiet assert gone.last == at(20, 9, 1) assert "последнее в 20.09 09:01" in brief(stopped) told(state, stopped) assert digest(state, swept([tail], at=at(21, 16))).quiet == () def test_a_trouble_that_returns_after_the_all_clear_is_new_again(tmp_path): state = seen(tmp_path) tail = burst("gitea", stamps(20, 9, 2)) told(state, digest(state, swept([tail], at=at(20, 10)))) told(state, digest(state, swept([tail], at=at(21, 12)))) # стихла back = digest( state, swept([burst("gitea", stamps(20, 9, 2) + stamps(21, 18, 3))], at=at(21, 19)), ) assert len(back.new) == 1 and back.worth_telling def test_the_background_of_the_baseline_never_claims_to_have_stopped(tmp_path): # Про фон, принятый в базовую линию, мастер не слышал - «стихло» про него врёт. state = seen(tmp_path, known=()) state.fresh = False cold = digest(state, swept([burst("lodge", stamps(20, 9, 2))], at=at(20, 10))) assert cold.new == () and cold.baseline == 1 gone = digest(state, swept([read_stack("lodge", log())], at=at(20, 13))) assert gone.quiet == () and not gone.worth_telling def test_the_summary_folds_the_variants_of_one_service_into_one_row(tmp_path): state = seen(tmp_path) # Разные по смыслу строки: отличие в одних цифрах - это одна и та же беда. variants = [ burst("gitea", stamps(20, 9, 4), text=f"ERROR {what}") for what in ("push отклонён", "миграция не прошла", "ssh-ключ отвергнут") ] told(state, digest(state, swept(variants, at=at(20, 10)))) night = digest(state, swept(variants, at=at(21, 1, 40))) (row,) = grouped(night.ongoing) assert row.kinds == 3 and row.day_before == 12 text = daily(night, UTC) assert "Идёт до сих пор: 1 рядов (3 разных строк)" in text assert "gitea/svc ошибка, строк 3 видов, с 20.09 09:00, за сутки 12" in text