refactor(smoke): срезаны проверки стенда, не служащие менти

- Зачем:
  - проверок стало больше, чем продукта, и росли они из критериев приёмки,
    а не из учебной ценности (#56).
- Что:
  - удалены tests/dag-probes-unit.py, tests/stand-smoke-guards.sh,
    tests/stand-smoke-static.sh и tests/smoke-guards.sh вместе с целью
    make smoke-guards и запуском юнит-тестов в scripts/config-test.sh.
  - из scripts/stand-smoke.sh убран check_env_consistency, туда же переехал
    check_keeper_runtime; счёт проверок остался 25.
  - в пробниках свёрнуты функции _assert_*, комментарий про отложенный импорт
    переписан на причину из документации Airflow и продублирован в test_kafka.
- Проверка:
  - make config-test, make up, make smoke, make smoke-cluster.

dags/test_clickhouse.py

@@ -42,9 +42,10 @@ NODES = (
 
 
 def _clickhouse_client():
-    # clickhouse_connect стоит только в образе Airflow, а малые проверки грузят
-    # этот модуль обычным интерпретатором, где пакета нет. Импорт верхнего
-    # уровня красит make config-test, поэтому он живёт здесь.
+    # clickhouse_connect импортируется внутри функции, а не наверху файла:
+    # обработчик DAG разбирает этот файл снова и снова, и импорт наверху
+    # оплачивался бы каждым разбором. Тяжёлые импорты Airflow советует
+    # держать внутри задач.
     import clickhouse_connect
 
     connection = Connection.get("clickhouse_default")
@@ -82,6 +83,8 @@ def _drop_tables(client) -> None:
     )
 
 
+# Единственная проверка, вынесенная из задач: её делают обе, до создания таблиц
+# и после уборки.
 def _assert_tables_absent(client) -> None:
     for node_name, source in NODES:
         remaining = _table_engines(client, source)
@@ -91,32 +94,6 @@ def _assert_tables_absent(client) -> None:
             )
 
 
-def _assert_tables_created(client) -> None:
-    for node_name, source in NODES:
-        actual_tables = _table_engines(client, source)
-        if actual_tables != EXPECTED_TABLES:
-            raise RuntimeError(
-                f"неверный набор таблиц на {node_name}: {actual_tables}"
-            )
-
-
-def _assert_marker_path(
-    *,
-    distributed_rows: list[tuple[int, str, str]],
-    write_hostname: str,
-    node_2_hostname: str,
-    marker: str,
-) -> None:
-    if write_hostname == node_2_hostname:
-        raise RuntimeError("запись и чтение маркера должны выполняться с разных нод")
-    expected_rows = [(1, write_hostname, marker)]
-    if distributed_rows != expected_rows:
-        raise RuntimeError(
-            "нода 2 не прочитала маркер первого шарда через Distributed: "
-            f"{marker}, получено {distributed_rows}"
-        )
-
-
 @dag(
     dag_id="test_clickhouse",
     schedule=None,
@@ -157,7 +134,12 @@ def test_clickhouse():
                 )
                 """
             )
-            _assert_tables_created(client)
+            for node_name, source in NODES:
+                actual_tables = _table_engines(client, source)
+                if actual_tables != EXPECTED_TABLES:
+                    raise RuntimeError(
+                        f"неверный набор таблиц на {node_name}: {actual_tables}"
+                    )
         finally:
             client.close()
 
@@ -211,12 +193,15 @@ def test_clickhouse():
                 """,
                 parameters={"marker": written["marker"]},
             ).result_rows
-            _assert_marker_path(
-                distributed_rows=distributed_rows,
-                write_hostname=written["hostname"],
-                node_2_hostname=node_2_rows[0][0],
-                marker=written["marker"],
-            )
+            if written["hostname"] == node_2_rows[0][0]:
+                raise RuntimeError(
+                    "запись и чтение маркера должны выполняться с разных нод"
+                )
+            if distributed_rows != [(1, written["hostname"], written["marker"])]:
+                raise RuntimeError(
+                    "нода 2 не прочитала маркер первого шарда через Distributed: "
+                    f"{written['marker']}, получено {distributed_rows}"
+                )
         finally:
             client.close()
 

dags/test_kafka.py

@@ -74,48 +74,6 @@ class RecordAddress(NamedTuple):
     offset: int
 
 
-def _assert_all_delivered(undelivered: int, marker: str) -> None:
-    if undelivered:
-        raise RuntimeError(
-            f"Kafka не приняла маркер за {FLUSH_TIMEOUT_SEC} с, "
-            f"не доставлено сообщений {undelivered}: {marker}"
-        )
-
-
-def _assert_no_delivery_errors(delivery_errors: list[str], marker: str) -> None:
-    if delivery_errors:
-        raise RuntimeError(
-            f"Kafka отказалась принять маркер {marker}: {delivery_errors}"
-        )
-
-
-def _assert_confirmed_once(addresses: list[RecordAddress], marker: str) -> None:
-    if len(addresses) != 1:
-        raise RuntimeError(
-            f"Kafka подтвердила запись маркера {marker} "
-            f"не одним сообщением: {addresses}"
-        )
-
-
-def _assert_message_arrived(message, marker: str) -> None:
-    if message is None:
-        raise RuntimeError(
-            f"Kafka молчала {READ_DEADLINE_SEC} с и не вернула маркер: {marker}"
-        )
-
-
-def _assert_no_read_error(message) -> None:
-    if message.error():
-        raise RuntimeError(f"Kafka вернула ошибку чтения: {message.error()}")
-
-
-def _assert_marker_matches(message, marker: str) -> None:
-    if message.value() != marker.encode():
-        raise RuntimeError(
-            f"по адресу записи лежит не маркер запуска {marker}: {message.value()!r}"
-        )
-
-
 @dag(
     dag_id="test_kafka",
     schedule=None,
@@ -127,6 +85,10 @@ def _assert_marker_matches(message, marker: str) -> None:
 def test_kafka():
     @task
     def write_marker() -> dict[str, str | int]:
+        # confluent_kafka импортируется внутри задачи, а не наверху файла:
+        # обработчик DAG разбирает этот файл снова и снова, и импорт наверху
+        # оплачивался бы каждым разбором. Тяжёлые импорты Airflow советует
+        # держать внутри задач.
         from confluent_kafka import Producer
 
         marker = f"{get_current_context()['run_id']}:{uuid.uuid4()}"
@@ -158,9 +120,20 @@ def test_kafka():
         finally:
             undelivered = producer.flush(FLUSH_TIMEOUT_SEC)
 
-        _assert_all_delivered(undelivered, marker)
-        _assert_no_delivery_errors(delivery_errors, marker)
-        _assert_confirmed_once(addresses, marker)
+        if undelivered:
+            raise RuntimeError(
+                f"Kafka не приняла маркер за {FLUSH_TIMEOUT_SEC} с, "
+                f"не доставлено сообщений {undelivered}: {marker}"
+            )
+        if delivery_errors:
+            raise RuntimeError(
+                f"Kafka отказалась принять маркер {marker}: {delivery_errors}"
+            )
+        if len(addresses) != 1:
+            raise RuntimeError(
+                f"Kafka подтвердила запись маркера {marker} "
+                f"не одним сообщением: {addresses}"
+            )
         return {
             "marker": marker,
             "partition": addresses[0].partition,
@@ -190,9 +163,17 @@ def test_kafka():
             # другого нечем, поэтому у чтения обязан быть крайний срок.
             while message is None and time.monotonic() < deadline:
                 message = consumer.poll(POLL_TIMEOUT_SEC)
-            _assert_message_arrived(message, marker)
-            _assert_no_read_error(message)
-            _assert_marker_matches(message, marker)
+            if message is None:
+                raise RuntimeError(
+                    f"Kafka молчала {READ_DEADLINE_SEC} с и не вернула маркер: {marker}"
+                )
+            if message.error():
+                raise RuntimeError(f"Kafka вернула ошибку чтения: {message.error()}")
+            if message.value() != marker.encode():
+                raise RuntimeError(
+                    f"по адресу записи лежит не маркер запуска {marker}: "
+                    f"{message.value()!r}"
+                )
         finally:
             consumer.close()