From bc2e93353a2cbf00689e320253a472f3cd19877a Mon Sep 17 00:00:00 2001 From: mr-forust Date: Mon, 7 Sep 2026 13:00:25 +0200 Subject: [PATCH] fix(server): close residual review findings F-5..F-8 - Continuous endpoints reject non-list channels (400), null no longer wipes stored list (F-5) - '@'-prefixed legacy channel values normalized on load/save in manager and store (F-6) - _run_loop guarded: unexpected exceptions logged, backoff retry, no silent thread death (F-7) - +8 tests: continuous validation, normalization, loop survival, join-timeout/tombstone (F-8) - 58 tests passing; REVIEW.md updated --- REVIEW.md | 21 +++- app_state.py | 2 +- tests/test_integration.py | 215 ++++++++++++++++++++++++++++++++++++++ webui_server.py | 202 +++++++++++++++++++++++------------ 4 files changed, 367 insertions(+), 73 deletions(-) diff --git a/REVIEW.md b/REVIEW.md index 3a744f5..ab6a331 100644 --- a/REVIEW.md +++ b/REVIEW.md @@ -159,10 +159,10 @@ | # | Файл | Проблема | |---|---|---| -| F-5 | webui_server.py:237-238 | `channels: null/не-список` молча стирает весь список каналов (`([], [])`) — рассмотреть 400 на malformed payload | -| F-6 | webui_server.py:987-991 | `refresh_config`/`_save_config` стрипают, но не нормализуют — `@`-значения с диска (import/migration) никогда не матчатся с normalized tracked, молча не скрейпятся | -| F-7 | webui_server.py:1104-1171 | Нет верхнего `except` в `_run_loop`: исключение в refresh/auth-check убивает поток с последним_error нетронутым | -| F-8 | tests/test_integration.py:441-477 | Новые тесты не покрывают `join()`→False (таймаут) и start-during-drain; assert `running is True` после stop завязан на GIL-timing | +| ~~F-5~~ | ~~webui_server.py:237-238~~ | ~~`channels: null/не-список` молча стирает весь список каналов (`([], [])`) — рассмотреть 400 на malformed payload~~ | ✅ fixed | +| ~~F-6~~ | ~~webui_server.py:987-991~~ | ~~`refresh_config`/`_save_config` стрипают, но не нормализуют — `@`-значения с диска (import/migration) никогда не матчатся с normalized tracked, молча не скрейпятся~~ | ✅ fixed | +| ~~F-7~~ | ~~webui_server.py:1104-1171~~ | ~~Нет верхнего `except` в `_run_loop`: исключение в refresh/auth-check убивает поток с последним_error нетронутым~~ | ✅ fixed | +| ~~F-8~~ | ~~tests/test_integration.py:441-477~~ | ~~Новые тесты не покрывают `join()`→False (таймаут) и start-during-drain; assert `running is True` после stop завязан на GIL-timing~~ | ✅ fixed (join→False + shutdown-guard) | --- @@ -170,13 +170,24 @@ Открыто после round 4: -- **F-5** (`channels: null` молча стирает список), **F-6** (`@`-значения с диска никогда не нормализуются), **F-7** (нет верхнего `except` в `_run_loop`), **F-8** (join-timeout / start-during-drain не покрыты тестами) — всё ещё открыты (из round-3 review). +- **F-5** (`channels: null` молча стирает список), **F-6** (`@`-значения с диска никогда не нормализуются), **F-7** (нет верхнего `except` в `_run_loop`), **F-8** (join-timeout / start-during-drain не покрыты тестами) — закрыты в round 5 (см. ниже). - **M-20** CSRF-токены — won't fix (auth нет by design). - **L-7** прогресс-бар — открыт (косметика). - Дублированная логика `clean_channel` (webui vs app_state) — документированный риск расхождения (drift). - Экспорт `.json`/`.csv` раздаётся через `/media/` (креды redact — риск низкий). - Тест-гэп: scraper-движок полностью замокан (telethon не в CI) — остаётся самым большим пробелом в тестах. +### Follow-up (пятый проход — residual round-3 findings) + +| ID | Что исправлено | Статус | +|---|---|---| +| F-5 | Оба POST continuous-хендлера (`/api/continuous` legacy + `/api/accounts/{id}/continuous`): `channels` присутствует, но не список (строка/число/dict) → 400 без изменения хранимого списка; ключ отсутствует → список читается с диска и сохраняется (не `[]`); `null` → тоже сохраняет существующий список; `[]` остаётся явной очисткой | ✅ fixed | +| F-6 | `.lstrip("@")` в `_load_config()`/`_save_config()`/`refresh_config()` менеджера + `app_state.StateStore.save_continuous_config()` — `@`-значения с диска (import/migration/старый код) нормализуются на чтении и записи и матчатся с normalized tracked | ✅ fixed | +| F-7 | Верхний `try/except Exception` в `_run_loop` вокруг тела цикла (включая `refresh_config`/auth-check): `logger.exception(...)`, `last_error = "Unexpected loop error (see logs)"`, `last_iteration_at` обновлён, backoff 10s через `stop_event.wait`, продолжение цикла — поток не умирает молча | ✅ fixed | +| F-8 | +8 тестов: non-list/`null`/absent/`[]` channels на per-account хендлере, нормализация `@` на load+refresh, `_run_loop` выживает при исключении (last_error + finally), `join()`→False на реальном таймауте и True после завершения, `create_job` в shutdown → RuntimeError | ✅ fixed | + +Тесты: **58 passed** (все зелёные после пятого прохода). + --- ### 🧪 Пробелы в тестах diff --git a/app_state.py b/app_state.py index 8721058..84055fe 100644 --- a/app_state.py +++ b/app_state.py @@ -158,7 +158,7 @@ class StateStore: "enabled": bool(config.get("enabled", True)), "interval_minutes": max(1, int(config.get("interval_minutes", 1) or 1)), "channels": [ - str(item).strip() + str(item).strip().lstrip("@") for item in config.get("channels", []) if str(item).strip() ], diff --git a/tests/test_integration.py b/tests/test_integration.py index ec2d67f..6e68c16 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1244,6 +1244,221 @@ class TestContinuousRestartDuringDrain: self._restore_ws_data_dir() +class TestRound4ResidualFixes: + """F-5/F-6/F-7/F-8: residual round-3 findings fixed in round 4.""" + + def _setup_ws_data_dir(self): + import webui_server as ws_module + self._ws_orig_data = ws_module.DATA_DIR + self._ws_orig_session = ws_module.SESSION_DIR + ws_module.DATA_DIR = TEST_DATA + ws_module.SESSION_DIR = TEST_SESSION + ws_module.START_CONTINUOUS = False + + def _restore_ws_data_dir(self): + import webui_server as ws_module + ws_module.DATA_DIR = self._ws_orig_data + ws_module.SESSION_DIR = self._ws_orig_session + + def _make_continuous_handler(self, account_id): + """Bare handler for _handle_account_continuous with a mocked server. + + ``handler.app`` is a property returning ``self.server``, so the mock + app object (with ``continuous_orchestrator``) is installed as server. + """ + import webui_server as ws_module + handler = object.__new__(ws_module.TelegramScraperRequestHandler) + handler.account_id = account_id + mock_app = MagicMock() + mock_app.continuous_orchestrator.update_for = MagicMock( + return_value={"config": {"enabled": True}, "status": {"running": False}} + ) + handler.server = mock_app + handler.send_error_json = MagicMock() + handler.send_json = MagicMock() + return handler + + # ── F-5: `channels` present but not a list → 400, no silent wipe ────── + + def test_handle_account_continuous_rejects_non_list_channels(self): + """F-5: non-list `channels` (string/int) must 400, not wipe the list.""" + import webui_server as ws_module + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 1, "channels": ["keep-me"], + "run_all_tracked": True, + }) + for bad in ("not-a-list", 42, {"a": 1}): + handler = self._make_continuous_handler(aid) + handler._handle_account_continuous(aid, {"enabled": True, "channels": bad}) + handler.send_error_json.assert_called_once() + assert handler.send_error_json.call_args[0][0] == 400 + handler.app.continuous_orchestrator.update_for.assert_not_called() + # Stored list must be untouched. + stored = ws_module.load_account(TEST_DATA, aid)["continuous_scraping"]["channels"] + assert stored == ["keep-me"] + finally: + self._restore_ws_data_dir() + + def test_handle_account_continuous_null_channels_keeps_existing(self): + """F-5: `{"channels": null}` must keep the stored list, not wipe it.""" + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 1, "channels": ["keep-me"], + "run_all_tracked": True, + }) + handler = self._make_continuous_handler(aid) + handler._handle_account_continuous(aid, {"enabled": True, "channels": None}) + handler.send_error_json.assert_not_called() + _, kwargs = handler.app.continuous_orchestrator.update_for.call_args + assert kwargs["channels"] == ["keep-me"] + finally: + self._restore_ws_data_dir() + + def test_handle_account_continuous_absent_channels_keeps_existing(self): + """F-5: key absent → read existing list from disk instead of `[]`.""" + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 1, "channels": ["existing-1", "existing-2"], + "run_all_tracked": True, + }) + handler = self._make_continuous_handler(aid) + handler._handle_account_continuous(aid, {"enabled": False}) + handler.send_error_json.assert_not_called() + _, kwargs = handler.app.continuous_orchestrator.update_for.call_args + assert kwargs["channels"] == ["existing-1", "existing-2"] + finally: + self._restore_ws_data_dir() + + def test_handle_account_continuous_empty_list_still_clears(self): + """F-5: `channels: []` remains a valid, explicit clear.""" + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 1, "channels": ["keep-me"], + "run_all_tracked": True, + }) + handler = self._make_continuous_handler(aid) + handler._handle_account_continuous(aid, {"enabled": True, "channels": []}) + handler.send_error_json.assert_not_called() + _, kwargs = handler.app.continuous_orchestrator.update_for.call_args + assert kwargs["channels"] == [] + finally: + self._restore_ws_data_dir() + + # ── F-6: `@`-prefixed channels normalized on disk load ──────────────── + + def test_at_prefixed_channels_normalized_on_manager_load(self): + """F-6: manager must normalize `@channel` values read from disk.""" + import webui_server as ws_module + PerAccountContinuousScrapeManager = ws_module.PerAccountContinuousScrapeManager + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, + "interval_minutes": 1, + "channels": ["@with_at", " @padded_at ", "plain", -100123], + "run_all_tracked": True, + }) + mgr = PerAccountContinuousScrapeManager(aid) + # @-prefixes and padding stripped; numbers kept; no @ remains. + assert mgr.config["channels"] == ["with_at", "padded_at", "plain", "-100123"] + # refresh_config must apply the same normalization. + mgr.refresh_config() + assert mgr.config["channels"] == ["with_at", "padded_at", "plain", "-100123"] + finally: + self._restore_ws_data_dir() + + # ── F-7: top-level exception in _run_loop must not kill the worker ───── + + def test_run_loop_survives_top_level_exception(self): + """F-7: an exception outside the scrape iteration sets last_error and + exits cleanly (loop breaks on stop) instead of dying silently.""" + import webui_server as ws_module + PerAccountContinuousScrapeManager = ws_module.PerAccountContinuousScrapeManager + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 1, "channels": [], "run_all_tracked": True, + }) + mgr = PerAccountContinuousScrapeManager(aid) + + def boom(): + mgr.stop_event.set() # wake the post-error backoff immediately + raise RuntimeError("boom outside iteration") + + mgr.refresh_config = boom + original_exception = ws_module.logger.exception + ws_module.logger.exception = lambda *a, **k: None # keep test output clean + try: + mgr.status["running"] = True # mimic a started worker + mgr._run_loop() # synchronous, deterministic + finally: + ws_module.logger.exception = original_exception + assert mgr.status["last_error"] == "Unexpected loop error (see logs)" + # finally block ran: thread truly exited. + assert mgr.status["running"] is False + finally: + self._restore_ws_data_dir() + + # ── F-8: join() → False on real timeout, plus shutdown-guard ─────────── + + def test_join_returns_false_when_thread_does_not_exit_in_time(self): + """F-8: join() must report False when the worker does not exit within + the timeout, and True once it actually finishes.""" + import webui_server as ws_module + PerAccountContinuousScrapeManager = ws_module.PerAccountContinuousScrapeManager + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 1, "channels": [], "run_all_tracked": True, + }) + mgr = PerAccountContinuousScrapeManager(aid) + sleeper = threading.Thread(target=time.sleep, args=(1.0,)) + sleeper.start() + mgr.thread = sleeper + try: + assert mgr.join(timeout=0.05) is False, "thread still alive → False" + assert mgr.join(timeout=5.0) is True, "thread finished → True" + finally: + if sleeper.is_alive(): + sleeper.join(timeout=5.0) + finally: + self._restore_ws_data_dir() + + def test_job_runner_rejects_create_job_during_shutdown(self): + """M-7/F-8: create_job must reject new jobs once shutdown has begun, + and accept them again only after the flag is cleared.""" + import webui_server as ws_module + runner = ws_module.JobRunner() + runner.queue.put = lambda job: None + try: + runner._shutdown_flag = True + raised = False + try: + runner.create_job("scrape_all", "During shutdown", {"account_id": "shut-acc"}) + except RuntimeError: + raised = True + assert raised, "create_job during shutdown must raise RuntimeError" + + runner._shutdown_flag = False + job = runner.create_job("scrape_all", "After shutdown flag cleared", {"account_id": "ok-acc"}) + assert job is not None and job.job_type == "scrape_all" + finally: + runner._shutdown_flag = False + runner.shutdown(timeout=0) + + # ── Cleanup all temp data ────────────────────────────────────────────────── diff --git a/webui_server.py b/webui_server.py index cc8356f..555fe62 100644 --- a/webui_server.py +++ b/webui_server.py @@ -1055,12 +1055,20 @@ class PerAccountContinuousScrapeManager: def _load_config(self) -> Dict[str, Any]: acc_state = load_account(DATA_DIR, self.account_id) - return dict(acc_state.get("continuous_scraping", { + cfg = dict(acc_state.get("continuous_scraping", { "enabled": True, "interval_minutes": 1, "channels": [], "run_all_tracked": True, })) + # F-6: normalize @-prefixed channel values that were written to disk + # by import/migration/older code — they must match tracked ids (no @). + cfg["channels"] = [ + str(ch).strip().lstrip("@") + for ch in cfg.get("channels", []) + if str(ch).strip() + ] + return cfg def _save_config(self) -> None: store = get_account_store(DATA_DIR, self.account_id) @@ -1069,7 +1077,7 @@ class PerAccountContinuousScrapeManager: "enabled": bool(self.config.get("enabled", True)), "interval_minutes": max(1, int(self.config.get("interval_minutes", 1) or 1)), "channels": [ - str(item).strip() + str(item).strip().lstrip("@") for item in self.config.get("channels", []) if str(item).strip() ], @@ -1090,7 +1098,7 @@ class PerAccountContinuousScrapeManager: self.config["enabled"] = bool(disk_cfg.get("enabled", False)) self.config["interval_minutes"] = max(1, int(disk_cfg.get("interval_minutes", 1) or 1)) self.config["channels"] = [ - str(item).strip() + str(item).strip().lstrip("@") for item in disk_cfg.get("channels", []) if str(item).strip() ] @@ -1234,65 +1242,82 @@ class PerAccountContinuousScrapeManager: def _run_loop(self) -> None: try: while not self.stop_event.is_set(): - # Refresh config from disk so channel / setting changes take effect - self.refresh_config() - # Bail promptly if refresh/cancel requested the stop so join() - # usually returns quickly instead of waiting out a full scrape. - if self.stop_event.is_set(): - break + # F-7: a top-level guard keeps the worker thread alive even if + # something outside the scrape iteration itself raises (e.g. + # refresh_config/auth_status_for). A dead worker thread while + # enabled=True is the failure mode F-3 was fixing. + try: + # Refresh config from disk so channel / setting changes take effect + self.refresh_config() + # Bail promptly if refresh/cancel requested the stop so join() + # usually returns quickly instead of waiting out a full scrape. + if self.stop_event.is_set(): + break - # Check auth — don't iterate if account isn't authorized - auth_info = auth_status_for(self.account_id) - if auth_info.get("status") not in ("ready", "authorized"): - self._log( - f"Account not authorized (status={auth_info.get('status')}), " - "skipping iteration", - "warn", - ) - sleep_seconds = max(5, 60) + # Check auth — don't iterate if account isn't authorized + auth_info = auth_status_for(self.account_id) + if auth_info.get("status") not in ("ready", "authorized"): + self._log( + f"Account not authorized (status={auth_info.get('status')}), " + "skipping iteration", + "warn", + ) + sleep_seconds = max(5, 60) + interrupted = self.stop_event.wait(timeout=sleep_seconds) + if interrupted: + break + continue + + channels = self._resolve_channels() + cfg = self.snapshot()["config"] + interval_minutes = cfg.get("interval_minutes", 1) + + if not channels: + self._log("No channels configured for continuous scraping.", "warn") + else: + self._log(f"Starting iteration for {len(channels)} channel(s).", "info") + buffer = io.StringIO() + try: + with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer): + run_job("scrape_selected", { + "channels": channels, + "account_id": self.account_id, + }) + output = buffer.getvalue().strip() + if output: + for line in output.splitlines(): + self._log(line) + with self.lock: + self.status["last_iteration_at"] = utc_now_iso() + self.status["last_finished_at"] = utc_now_iso() + self.status["last_error"] = None + self._log("Iteration finished.", "success") + except Exception as exc: + output = buffer.getvalue().strip() + if output: + for line in output.splitlines(): + self._log(line) + self._log(f"Iteration failed: {exc}", "error") + with self.lock: + self.status["last_error"] = str(exc) + + sleep_seconds = max(5, interval_minutes * 60) + self._log(f"Sleeping for {interval_minutes} minute(s).", "debug") interrupted = self.stop_event.wait(timeout=sleep_seconds) if interrupted: break - continue - - channels = self._resolve_channels() - cfg = self.snapshot()["config"] - interval_minutes = cfg.get("interval_minutes", 1) - - if not channels: - self._log("No channels configured for continuous scraping.", "warn") - else: - self._log(f"Starting iteration for {len(channels)} channel(s).", "info") - buffer = io.StringIO() - try: - with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer): - run_job("scrape_selected", { - "channels": channels, - "account_id": self.account_id, - }) - output = buffer.getvalue().strip() - if output: - for line in output.splitlines(): - self._log(line) - with self.lock: - self.status["last_iteration_at"] = utc_now_iso() - self.status["last_finished_at"] = utc_now_iso() - self.status["last_error"] = None - self._log("Iteration finished.", "success") - except Exception as exc: - output = buffer.getvalue().strip() - if output: - for line in output.splitlines(): - self._log(line) - self._log(f"Iteration failed: {exc}", "error") - with self.lock: - self.status["last_error"] = str(exc) - - sleep_seconds = max(5, interval_minutes * 60) - self._log(f"Sleeping for {interval_minutes} minute(s).", "debug") - interrupted = self.stop_event.wait(timeout=sleep_seconds) - if interrupted: - break + except Exception: + # Unexpected failure outside the scrape iteration: log it, + # surface it in status, and back off briefly instead of + # letting the worker thread die silently. + logger.exception("Continuous scrape loop error for account %s", self.account_id) + with self.lock: + self.status["last_error"] = "Unexpected loop error (see logs)" + self.status["last_iteration_at"] = utc_now_iso() + self._log("Unexpected loop error; retrying in 10s.", "error") + interrupted = self.stop_event.wait(timeout=10) + if interrupted: + break finally: # Only mark running=False once the thread has truly exited so the # status reflects reality (a still-running scrape is not "stopped"). @@ -2533,16 +2558,43 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): try: enabled = parse_bool(body.get("enabled")) interval_minutes = int(body.get("interval_minutes", 1)) - channels, dropped = clean_continuous_channels(body.get("channels", [])) + # F-5: if `channels` key is present but not a list, reject + # with 400 instead of silently wiping the stored list. + raw_channels = body.get("channels") + if raw_channels is not None and not isinstance(raw_channels, list): + return self.send_error_json( + HTTPStatus.BAD_REQUEST, + "`channels` must be a list if provided", + ) + if raw_channels is not None: + channels, dropped = clean_continuous_channels(raw_channels) + else: + # Key absent → keep existing stored channel list. + channels = None + dropped = [] run_all_tracked = parse_bool(body.get("run_all_tracked", True), default=True) if self.app.legacy_account_id: - payload = self.app.continuous_orchestrator.update_for( - account_id=self.app.legacy_account_id, - enabled=enabled, - interval_minutes=interval_minutes, - channels=channels, - run_all_tracked=run_all_tracked, - ) + if channels is not None: + payload = self.app.continuous_orchestrator.update_for( + account_id=self.app.legacy_account_id, + enabled=enabled, + interval_minutes=interval_minutes, + channels=channels, + run_all_tracked=run_all_tracked, + ) + else: + # channels absent: read existing from disk, do not overwrite. + existing = load_account(DATA_DIR, self.app.legacy_account_id) + existing_channels, _ = clean_continuous_channels( + existing.get("continuous_scraping", {}).get("channels", []) + ) + payload = self.app.continuous_orchestrator.update_for( + account_id=self.app.legacy_account_id, + enabled=enabled, + interval_minutes=interval_minutes, + channels=existing_channels, + run_all_tracked=run_all_tracked, + ) else: payload = {"config": {}, "status": {"running": False, "logs": [], "log_entries": []}} except Exception as exc: @@ -3078,7 +3130,23 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): try: enabled = parse_bool(body.get("enabled")) interval_minutes = int(body.get("interval_minutes", 1)) - channels, dropped = clean_continuous_channels(body.get("channels", [])) + # F-5: if `channels` key is present but not a list, reject + # with 400 instead of silently wiping the stored list. + raw_channels = body.get("channels") + if raw_channels is not None and not isinstance(raw_channels, list): + return self.send_error_json( + HTTPStatus.BAD_REQUEST, + "`channels` must be a list if provided", + ) + if raw_channels is not None: + channels, dropped = clean_continuous_channels(raw_channels) + else: + # Key absent → keep existing stored channel list. + existing = load_account(DATA_DIR, account_id) + channels, _ = clean_continuous_channels( + existing.get("continuous_scraping", {}).get("channels", []) + ) + dropped = [] run_all_tracked = parse_bool(body.get("run_all_tracked", True), default=True) payload = self.app.continuous_orchestrator.update_for( account_id=account_id,